<!doctype html><html lang=zh class=no-js> <head><meta charset=utf-8><meta name=viewport content="width=device-width,initial-scale=1"><link rel="shortcut icon" href=../../assets/images/favicon.png><meta name=generator content="mkdocs-1.1.2, mkdocs-material-5.5.13"><title>SparkCore - Dayet</title><link rel=stylesheet href=../../assets/stylesheets/main.077507d7.min.css><link rel=stylesheet href=../../assets/stylesheets/palette.ff0a5ce4.min.css><meta name=theme-color content=#546d78><link href=https://fonts.gstatic.com rel=preconnect crossorigin><link rel=stylesheet href="https://fonts.googleapis.com/css?family=Roboto:300,400,400i,700%7CConsolas&display=fallback"><style>body,input{font-family:"Roboto",-apple-system,BlinkMacSystemFont,Helvetica,Arial,sans-serif}code,kbd,pre{font-family:"Consolas",SFMono-Regular,Consolas,Menlo,monospace}</style><script>window.ga=window.ga||function(){(ga.q=ga.q||[]).push(arguments)},ga.l=+new Date,ga("create","UA-XXXXXXXX-X","auto"),ga("set","anonymizeIp",!0),ga("send","pageview"),document.addEventListener("DOMContentLoaded",function(){document.forms.search&&document.forms.search.query.addEventListener("blur",function(){if(this.value){var e=document.location.pathname;ga("send","pageview",e+"?q="+this.value)}})}),document.addEventListener("DOMContentSwitch",function(){ga("send","pageview",document.location.pathname)})</script><script async src=https://www.google-analytics.com/analytics.js></script></head> <body dir=ltr data-md-color-scheme data-md-color-primary=blue-grey data-md-color-accent=blue-grey> <input class=md-toggle data-md-toggle=drawer type=checkbox id=__drawer autocomplete=off> <input class=md-toggle data-md-toggle=search type=checkbox id=__search autocomplete=off> <label class=md-overlay for=__drawer></label> <div data-md-component=skip> <a href=#sparkcore class=md-skip> 跳转至 </a> </div> <div data-md-component=announce> </div> <header class=md-header data-md-component=header> <nav class="md-header-nav md-grid" aria-label=Header> <a href=../.. title=Dayet class="md-header-nav__button md-logo" aria-label=Dayet> <svg xmlns=http://www.w3.org/2000/svg viewbox="0 0 24 24"><path d="M12 8a3 3 0 003-3 3 3 0 00-3-3 3 3 0 00-3 3 3 3 0 003 3m0 3.54C9.64 9.35 6.5 8 3 8v11c3.5 0 6.64 1.35 9 3.54 2.36-2.19 5.5-3.54 9-3.54V8c-3.5 0-6.64 1.35-9 3.54z"/></svg> </a> <label class="md-header-nav__button md-icon" for=__drawer> <svg xmlns=http://www.w3.org/2000/svg viewbox="0 0 24 24"><path d="M3 6h18v2H3V6m0 5h18v2H3v-2m0 5h18v2H3v-2z"/></svg> </label> <div class=md-header-nav__title data-md-component=header-title> <div class=md-header-nav__ellipsis> <span class="md-header-nav__topic md-ellipsis"> Dayet </span> <span class="md-header-nav__topic md-ellipsis"> SparkCore </span> </div> </div> <label class="md-header-nav__button md-icon" for=__search> <svg xmlns=http://www.w3.org/2000/svg viewbox="0 0 24 24"><path d="M9.5 3A6.5 6.5 0 0116 9.5c0 1.61-.59 3.09-1.56 4.23l.27.27h.79l5 5-1.5 1.5-5-5v-.79l-.27-.27A6.516 6.516 0 019.5 16 6.5 6.5 0 013 9.5 6.5 6.5 0 019.5 3m0 2C7 5 5 7 5 9.5S7 14 9.5 14 14 12 14 9.5 12 5 9.5 5z"/></svg> </label> <div class=md-search data-md-component=search role=dialog> <label class=md-search__overlay for=__search></label> <div class=md-search__inner role=search> <form class=md-search__form name=search> <input type=text class=md-search__input name=query aria-label=搜索 placeholder=搜索 autocapitalize=off autocorrect=off autocomplete=off spellcheck=false data-md-component=search-query data-md-state=active> <label class="md-search__icon md-icon" for=__search> <svg xmlns=http://www.w3.org/2000/svg viewbox="0 0 24 24"><path d="M9.5 3A6.5 6.5 0 0116 9.5c0 1.61-.59 3.09-1.56 4.23l.27.27h.79l5 5-1.5 1.5-5-5v-.79l-.27-.27A6.516 6.516 0 019.5 16 6.5 6.5 0 013 9.5 6.5 6.5 0 019.5 3m0 2C7 5 5 7 5 9.5S7 14 9.5 14 14 12 14 9.5 12 5 9.5 5z"/></svg> <svg xmlns=http://www.w3.org/2000/svg viewbox="0 0 24 24"><path d="M20 11v2H8l5.5 5.5-1.42 1.42L4.16 12l7.92-7.92L13.5 5.5 8 11h12z"/></svg> </label> <button type=reset class="md-search__icon md-icon" aria-label=Clear data-md-component=search-reset tabindex=-1> <svg xmlns=http://www.w3.org/2000/svg viewbox="0 0 24 24"><path d="M19 6.41L17.59 5 12 10.59 6.41 5 5 6.41 10.59 12 5 17.59 6.41 19 12 13.41 17.59 19 19 17.59 13.41 12 19 6.41z"/></svg> </button> </form> <div class=md-search__output> <div class=md-search__scrollwrap data-md-scrollfix> <div class=md-search-result data-md-component=search-result> <div class=md-search-result__meta> Initializing search </div> <ol class=md-search-result__list></ol> </div> </div> </div> </div> </div> <div class=md-header-nav__source> <a href=https://chokgit.gitee.io/blog title="前往 GitHub 仓库" class=md-source> <div class="md-source__icon md-icon"> <svg xmlns=http://www.w3.org/2000/svg viewbox="0 0 448 512"><path d="M439.55 236.05L244 40.45a28.87 28.87 0 00-40.81 0l-40.66 40.63 51.52 51.52c27.06-9.14 52.68 16.77 43.39 43.68l49.66 49.66c34.23-11.8 61.18 31 35.47 56.69-26.49 26.49-70.21-2.87-56-37.34L240.22 199v121.85c25.3 12.54 22.26 41.85 9.08 55a34.34 34.34 0 01-48.55 0c-17.57-17.6-11.07-46.91 11.25-56v-123c-20.8-8.51-24.6-30.74-18.64-45L142.57 101 8.45 235.14a28.86 28.86 0 000 40.81l195.61 195.6a28.86 28.86 0 0040.8 0l194.69-194.69a28.86 28.86 0 000-40.81z"/></svg> </div> <div class=md-source__repository> 吾生也有崖，而知也无涯。 </div> </a> </div> </nav> </header> <div class=md-container data-md-component=container> <main class=md-main data-md-component=main> <div class="md-main__inner md-grid"> <div class="md-sidebar md-sidebar--primary" data-md-component=navigation> <div class=md-sidebar__scrollwrap> <div class=md-sidebar__inner> <nav class="md-nav md-nav--primary" aria-label=Navigation data-md-level=0> <label class=md-nav__title for=__drawer> <a href=../.. title=Dayet class="md-nav__button md-logo" aria-label=Dayet> <svg xmlns=http://www.w3.org/2000/svg viewbox="0 0 24 24"><path d="M12 8a3 3 0 003-3 3 3 0 00-3-3 3 3 0 00-3 3 3 3 0 003 3m0 3.54C9.64 9.35 6.5 8 3 8v11c3.5 0 6.64 1.35 9 3.54 2.36-2.19 5.5-3.54 9-3.54V8c-3.5 0-6.64 1.35-9 3.54z"/></svg> </a> Dayet </label> <div class=md-nav__source> <a href=https://chokgit.gitee.io/blog title="前往 GitHub 仓库" class=md-source> <div class="md-source__icon md-icon"> <svg xmlns=http://www.w3.org/2000/svg viewbox="0 0 448 512"><path d="M439.55 236.05L244 40.45a28.87 28.87 0 00-40.81 0l-40.66 40.63 51.52 51.52c27.06-9.14 52.68 16.77 43.39 43.68l49.66 49.66c34.23-11.8 61.18 31 35.47 56.69-26.49 26.49-70.21-2.87-56-37.34L240.22 199v121.85c25.3 12.54 22.26 41.85 9.08 55a34.34 34.34 0 01-48.55 0c-17.57-17.6-11.07-46.91 11.25-56v-123c-20.8-8.51-24.6-30.74-18.64-45L142.57 101 8.45 235.14a28.86 28.86 0 000 40.81l195.61 195.6a28.86 28.86 0 0040.8 0l194.69-194.69a28.86 28.86 0 000-40.81z"/></svg> </div> <div class=md-source__repository> 吾生也有崖，而知也无涯。 </div> </a> </div> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=../.. title=主页 class=md-nav__link> 主页 </a> </li> <li class="md-nav__item md-nav__item--nested"> <input class="md-nav__toggle md-toggle" data-md-toggle=nav-2 type=checkbox id=nav-2> <label class=md-nav__link for=nav-2> 系统基础 <span class="md-nav__icon md-icon"></span> </label> <nav class=md-nav aria-label=系统基础 data-md-level=1> <label class=md-nav__title for=nav-2> <span class="md-nav__icon md-icon"></span> 系统基础 </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=../../%E7%B3%BB%E7%BB%9F/Linux/ title=Linux class=md-nav__link> Linux </a> </li> <li class=md-nav__item> <a href=../../%E7%B3%BB%E7%BB%9F/Shell/ title=Shell class=md-nav__link> Shell </a> </li> </ul> </nav> </li> <li class="md-nav__item md-nav__item--nested"> <input class="md-nav__toggle md-toggle" data-md-toggle=nav-3 type=checkbox id=nav-3> <label class=md-nav__link for=nav-3> 编程语言 <span class="md-nav__icon md-icon"></span> </label> <nav class=md-nav aria-label=编程语言 data-md-level=1> <label class=md-nav__title for=nav-3> <span class="md-nav__icon md-icon"></span> 编程语言 </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=../../%E5%BF%85%E5%A4%87%E6%8A%80%E6%9C%AF/JavaSE/ title=Java class=md-nav__link> Java </a> </li> <li class=md-nav__item> <a href=../../%E5%BF%85%E5%A4%87%E6%8A%80%E6%9C%AF/Scala/ title=Scala class=md-nav__link> Scala </a> </li> <li class=md-nav__item> <a href=../../%E5%BF%85%E5%A4%87%E6%8A%80%E6%9C%AF/Python/ title=Python class=md-nav__link> Python </a> </li> <li class=md-nav__item> <a href=../../%E5%BF%85%E5%A4%87%E6%8A%80%E6%9C%AF/Kotlin/ title=Kotlin class=md-nav__link> Kotlin </a> </li> </ul> </nav> </li> <li class="md-nav__item md-nav__item--nested"> <input class="md-nav__toggle md-toggle" data-md-toggle=nav-4 type=checkbox id=nav-4> <label class=md-nav__link for=nav-4> 前端技术 <span class="md-nav__icon md-icon"></span> </label> <nav class=md-nav aria-label=前端技术 data-md-level=1> <label class=md-nav__title for=nav-4> <span class="md-nav__icon md-icon"></span> 前端技术 </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=../../%E5%89%8D%E7%AB%AF/Vue/ title=Vue class=md-nav__link> Vue </a> </li> <li class=md-nav__item> <a href=../../%E5%89%8D%E7%AB%AF/Element-UI/ title=Element-UI-admin class=md-nav__link> Element-UI-admin </a> </li> </ul> </nav> </li> <li class="md-nav__item md-nav__item--nested"> <input class="md-nav__toggle md-toggle" data-md-toggle=nav-5 type=checkbox id=nav-5> <label class=md-nav__link for=nav-5> 后端技术 <span class="md-nav__icon md-icon"></span> </label> <nav class=md-nav aria-label=后端技术 data-md-level=1> <label class=md-nav__title for=nav-5> <span class="md-nav__icon md-icon"></span> 后端技术 </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=../../%E5%90%8E%E7%AB%AF/SpringBoot/ title=SpringBoot class=md-nav__link> SpringBoot </a> </li> <li class=md-nav__item> <a href=../../%E5%90%8E%E7%AB%AF/Mybatis/ title=Mybatis class=md-nav__link> Mybatis </a> </li> <li class=md-nav__item> <a href=../../%E5%90%8E%E7%AB%AF/SpringData/ title=SpringData class=md-nav__link> SpringData </a> </li> </ul> </nav> </li> <li class="md-nav__item md-nav__item--nested"> <input class="md-nav__toggle md-toggle" data-md-toggle=nav-6 type=checkbox id=nav-6> <label class=md-nav__link for=nav-6> 中间件 <span class="md-nav__icon md-icon"></span> </label> <nav class=md-nav aria-label=中间件 data-md-level=1> <label class=md-nav__title for=nav-6> <span class="md-nav__icon md-icon"></span> 中间件 </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=../../%E4%B8%AD%E9%97%B4%E4%BB%B6/Nginx/ title=Nginx class=md-nav__link> Nginx </a> </li> <li class=md-nav__item> <a href=../../%E4%B8%AD%E9%97%B4%E4%BB%B6/Tomcat/ title=Tomcat class=md-nav__link> Tomcat </a> </li> <li class=md-nav__item> <a href=../../%E4%B8%AD%E9%97%B4%E4%BB%B6/RabbitMQ/ title=RabbitMQ class=md-nav__link> RabbitMQ </a> </li> <li class=md-nav__item> <a href=../../%E4%B8%AD%E9%97%B4%E4%BB%B6/RocketMQ/ title=RocketMQ class=md-nav__link> RocketMQ </a> </li> </ul> </nav> </li> <li class="md-nav__item md-nav__item--active md-nav__item--nested"> <input class="md-nav__toggle md-toggle" data-md-toggle=nav-7 type=checkbox id=nav-7 checked> <label class=md-nav__link for=nav-7> 大数据 <span class="md-nav__icon md-icon"></span> </label> <nav class=md-nav aria-label=大数据 data-md-level=1> <label class=md-nav__title for=nav-7> <span class="md-nav__icon md-icon"></span> 大数据 </label> <ul class=md-nav__list data-md-scrollfix> <li class="md-nav__item md-nav__item--nested"> <input class="md-nav__toggle md-toggle" data-md-toggle=nav-7-1 type=checkbox id=nav-7-1> <label class=md-nav__link for=nav-7-1> Hadoop <span class="md-nav__icon md-icon"></span> </label> <nav class=md-nav aria-label=Hadoop data-md-level=2> <label class=md-nav__title for=nav-7-1> <span class="md-nav__icon md-icon"></span> Hadoop </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=../Hadoop%E5%9F%BA%E7%A1%80/ title=入门 class=md-nav__link> 入门 </a> </li> <li class=md-nav__item> <a href=../Hadoop-HDFS/ title=HDFS class=md-nav__link> HDFS </a> </li> <li class=md-nav__item> <a href=../Hadoop-MapReduce/ title=MapReduce class=md-nav__link> MapReduce </a> </li> </ul> </nav> </li> <li class="md-nav__item md-nav__item--active md-nav__item--nested"> <input class="md-nav__toggle md-toggle" data-md-toggle=nav-7-2 type=checkbox id=nav-7-2 checked> <label class=md-nav__link for=nav-7-2> Spark <span class="md-nav__icon md-icon"></span> </label> <nav class=md-nav aria-label=Spark data-md-level=2> <label class=md-nav__title for=nav-7-2> <span class="md-nav__icon md-icon"></span> Spark </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=../Spark%E5%9F%BA%E7%A1%80/ title=Spark基础 class=md-nav__link> Spark基础 </a> </li> <li class="md-nav__item md-nav__item--active"> <input class="md-nav__toggle md-toggle" data-md-toggle=toc type=checkbox id=__toc> <label class="md-nav__link md-nav__link--active" for=__toc> SparkCore <span class="md-nav__icon md-icon"></span> </label> <a href=./ title=SparkCore class="md-nav__link md-nav__link--active"> SparkCore </a> <nav class="md-nav md-nav--secondary" aria-label=目录> <label class=md-nav__title for=__toc> <span class="md-nav__icon md-icon"></span> 目录 </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=#1-rdd class=md-nav__link> 第1章 RDD概述 </a> </li> <li class=md-nav__item> <a href=#11-rdd class=md-nav__link> 1.1 什么是RDD </a> </li> <li class=md-nav__item> <a href=#12-rdd class=md-nav__link> 1.2 RDD的属性 </a> </li> <li class=md-nav__item> <a href=#13-rdd class=md-nav__link> 1.3 RDD特点 </a> <nav class=md-nav aria-label="1.3 RDD特点"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#131 class=md-nav__link> 1.3.1 分区 </a> </li> <li class=md-nav__item> <a href=#132 class=md-nav__link> 1.3.2 只读 </a> </li> <li class=md-nav__item> <a href=#133 class=md-nav__link> 1.3.3 依赖 </a> </li> <li class=md-nav__item> <a href=#134 class=md-nav__link> 1.3.4 缓存 </a> </li> <li class=md-nav__item> <a href=#135-checkpoint class=md-nav__link> 1.3.5 CheckPoint </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#2-rdd class=md-nav__link> 第2章 RDD编程 </a> </li> <li class=md-nav__item> <a href=#21 class=md-nav__link> 2.1 编程模型 </a> </li> <li class=md-nav__item> <a href=#22-rdd class=md-nav__link> 2.2 RDD的创建 </a> <nav class=md-nav aria-label="2.2 RDD的创建"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#221 class=md-nav__link> 2.2.1 从集合中创建 </a> </li> <li class=md-nav__item> <a href=#222 class=md-nav__link> 2.2.2 由外部存储系统的数据集创建 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#23-rdd class=md-nav__link> 2.3 RDD的转换（面试开发重点） </a> <nav class=md-nav aria-label="2.3 RDD的转换（面试开发重点）"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#231-value class=md-nav__link> 2.3.1 Value类型 </a> <nav class=md-nav aria-label="2.3.1 Value类型"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#2311-mapfunc class=md-nav__link> 2.3.1.1 map(func)案例 </a> </li> <li class=md-nav__item> <a href=#2312-mappartitionsfunc class=md-nav__link> 2.3.1.2 mapPartitions(func) 案例 </a> </li> <li class=md-nav__item> <a href=#2313-mappartitionswithindexfunc class=md-nav__link> 2.3.1.3 mapPartitionsWithIndex(func) 案例 </a> </li> <li class=md-nav__item> <a href=#2314-flatmapfunc class=md-nav__link> 2.3.1.4 flatMap(func) 案例 </a> </li> <li class=md-nav__item> <a href=#2315-mapmappartition class=md-nav__link> 2.3.1.5 map()和mapPartition()的区别 </a> </li> <li class=md-nav__item> <a href=#2316-glom class=md-nav__link> 2.3.1.6 glom案例 </a> </li> <li class=md-nav__item> <a href=#2317-groupbyfunc class=md-nav__link> 2.3.1.7 groupBy(func)案例 </a> </li> <li class=md-nav__item> <a href=#2318-filterfunc class=md-nav__link> 2.3.1.8 filter(func) 案例 </a> </li> <li class=md-nav__item> <a href=#2319-samplewithreplacement-fraction-seed class=md-nav__link> 2.3.1.9 sample(withReplacement, fraction, seed) 案例 </a> </li> <li class=md-nav__item> <a href=#23110-distinctnumtasks class=md-nav__link> 2.3.1.10 distinct([numTasks])) 案例 </a> </li> <li class=md-nav__item> <a href=#23111-coalescenumpartitions class=md-nav__link> 2.3.1.11 coalesce(numPartitions) 案例 </a> </li> <li class=md-nav__item> <a href=#23112-repartitionnumpartitions class=md-nav__link> 2.3.1.12 repartition(numPartitions) 案例 </a> </li> <li class=md-nav__item> <a href=#23113-coalescerepartition class=md-nav__link> 2.3.1.13 coalesce和repartition的区别 </a> </li> <li class=md-nav__item> <a href=#23114-sortbyfuncascending-numtasks class=md-nav__link> 2.3.1.14 sortBy(func,[ascending], [numTasks]) 案例 </a> </li> <li class=md-nav__item> <a href=#23115-pipecommand-envvars class=md-nav__link> 2.3.1.15 pipe(command, [envVars]) 案例 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#232-value class=md-nav__link> 2.3.2 双Value类型交互 </a> <nav class=md-nav aria-label="2.3.2 双Value类型交互"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#2321-unionotherdataset class=md-nav__link> 2.3.2.1 union(otherDataset) 案例 </a> </li> <li class=md-nav__item> <a href=#2322-subtract-otherdataset class=md-nav__link> 2.3.2.2 subtract (otherDataset) 案例 </a> </li> <li class=md-nav__item> <a href=#2323-intersectionotherdataset class=md-nav__link> 2.3.2.3 intersection(otherDataset) 案例 </a> </li> <li class=md-nav__item> <a href=#2324-cartesianotherdataset class=md-nav__link> 2.3.2.4 cartesian(otherDataset) 案例 </a> </li> <li class=md-nav__item> <a href=#2325-zipotherdataset class=md-nav__link> 2.3.2.5 zip(otherDataset)案例 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#233-key-value class=md-nav__link> 2.3.3 Key-Value类型 </a> <nav class=md-nav aria-label="2.3.3 Key-Value类型"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#2331-partitionby class=md-nav__link> 2.3.3.1 partitionBy案例 </a> </li> <li class=md-nav__item> <a href=#2332-groupbykey class=md-nav__link> 2.3.3.2 groupByKey案例 </a> </li> <li class=md-nav__item> <a href=#2333-reducebykeyfunc-numtasks class=md-nav__link> 2.3.3.3 reduceByKey(func, [numTasks]) 案例 </a> </li> <li class=md-nav__item> <a href=#2334-reducebykeygroupbykey class=md-nav__link> 2.3.3.4 reduceByKey和groupByKey的区别 </a> </li> <li class=md-nav__item> <a href=#2335-aggregatebykey class=md-nav__link> 2.3.3.5 aggregateByKey案例 </a> </li> <li class=md-nav__item> <a href=#2336-foldbykey class=md-nav__link> 2.3.3.6 foldByKey案例 </a> </li> <li class=md-nav__item> <a href=#2337-combinebykeyc class=md-nav__link> 2.3.3.7 combineByKey[C] 案例 </a> </li> <li class=md-nav__item> <a href=#2338-sortbykeyascending-numtasks class=md-nav__link> 2.3.3.8 sortByKey([ascending], [numTasks]) 案例 </a> </li> <li class=md-nav__item> <a href=#2339-mapvalues class=md-nav__link> 2.3.3.9 mapValues案例 </a> </li> <li class=md-nav__item> <a href=#23310-joinotherdataset-numtasks class=md-nav__link> 2.3.3.10 join(otherDataset, [numTasks]) 案例 </a> </li> <li class=md-nav__item> <a href=#23311-cogroupotherdataset-numtasks class=md-nav__link> 2.3.3.11 cogroup(otherDataset, [numTasks]) 案例 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#234 class=md-nav__link> 2.3.4 案例实操 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#24-action class=md-nav__link> 2.4 Action </a> <nav class=md-nav aria-label="2.4 Action"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#241-reducefunc class=md-nav__link> 2.4.1 reduce(func)案例 </a> </li> <li class=md-nav__item> <a href=#242-collect class=md-nav__link> 2.4.2 collect()案例 </a> </li> <li class=md-nav__item> <a href=#243-count class=md-nav__link> 2.4.3 count()案例 </a> </li> <li class=md-nav__item> <a href=#244-first class=md-nav__link> 2.4.4 first()案例 </a> </li> <li class=md-nav__item> <a href=#245-taken class=md-nav__link> 2.4.5 take(n)案例 </a> </li> <li class=md-nav__item> <a href=#246-takeorderedn class=md-nav__link> 2.4.6 takeOrdered(n)案例 </a> </li> <li class=md-nav__item> <a href=#247-aggregate class=md-nav__link> 2.4.7 aggregate案例 </a> </li> <li class=md-nav__item> <a href=#248-foldnumfunc class=md-nav__link> 2.4.8 fold(num)(func)案例 </a> </li> <li class=md-nav__item> <a href=#249-saveastextfilepath class=md-nav__link> 2.4.9 saveAsTextFile(path) </a> </li> <li class=md-nav__item> <a href=#2410-saveassequencefilepath class=md-nav__link> 2.4.10 saveAsSequenceFile(path) </a> </li> <li class=md-nav__item> <a href=#2411-saveasobjectfilepath class=md-nav__link> 2.4.11 saveAsObjectFile(path) </a> </li> <li class=md-nav__item> <a href=#2412-countbykey class=md-nav__link> 2.4.12 countByKey()案例 </a> </li> <li class=md-nav__item> <a href=#2413-foreachfunc class=md-nav__link> 2.4.13 foreach(func)案例 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#25-rdd class=md-nav__link> 2.5 RDD中的函数传递 </a> <nav class=md-nav aria-label="2.5 RDD中的函数传递"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#251 class=md-nav__link> 2.5.1 传递一个方法 </a> </li> <li class=md-nav__item> <a href=#252 class=md-nav__link> 2.5.2 传递一个属性 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#26-rdd class=md-nav__link> 2.6 RDD依赖关系 </a> <nav class=md-nav aria-label="2.6 RDD依赖关系"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#261-lineage class=md-nav__link> 2.6.1 Lineage </a> </li> <li class=md-nav__item> <a href=#262 class=md-nav__link> 2.6.2 窄依赖 </a> </li> <li class=md-nav__item> <a href=#263 class=md-nav__link> 2.6.3 宽依赖 </a> </li> <li class=md-nav__item> <a href=#264-dag class=md-nav__link> 2.6.4 DAG </a> </li> <li class=md-nav__item> <a href=#265 class=md-nav__link> 2.6.5 任务划分（面试重点） </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#27-rdd class=md-nav__link> 2.7 RDD缓存 </a> </li> <li class=md-nav__item> <a href=#28-rdd-checkpoint class=md-nav__link> 2.8 RDD CheckPoint </a> </li> <li class=md-nav__item> <a href=#3-rdd class=md-nav__link> 第3章 键值对RDD数据分区器 </a> </li> <li class=md-nav__item> <a href=#31-rdd class=md-nav__link> 3.1 获取RDD分区 </a> </li> <li class=md-nav__item> <a href=#32-hash class=md-nav__link> 3.2 Hash分区 </a> </li> <li class=md-nav__item> <a href=#33-ranger class=md-nav__link> 3.3 Ranger分区 </a> </li> <li class=md-nav__item> <a href=#34 class=md-nav__link> 3.4 自定义分区 </a> </li> <li class=md-nav__item> <a href=#4 class=md-nav__link> 第4章 数据读取与保存 </a> </li> <li class=md-nav__item> <a href=#_1 class=md-nav__link> 文件类数据读取与保存 </a> <nav class=md-nav aria-label=文件类数据读取与保存> <ul class=md-nav__list> <li class=md-nav__item> <a href=#411-text class=md-nav__link> 4.1.1 Text文件 </a> </li> <li class=md-nav__item> <a href=#412-json class=md-nav__link> 4.1.2 Json文件 </a> </li> <li class=md-nav__item> <a href=#413-sequence class=md-nav__link> 4.1.3 Sequence文件 </a> </li> <li class=md-nav__item> <a href=#414 class=md-nav__link> 4.1.4 对象文件 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#_2 class=md-nav__link> 文件系统类数据读取与保存 </a> <nav class=md-nav aria-label=文件系统类数据读取与保存> <ul class=md-nav__list> <li class=md-nav__item> <a href=#421-hdfs class=md-nav__link> 4.2.1 HDFS </a> </li> <li class=md-nav__item> <a href=#422-mysql class=md-nav__link> 4.2.2 MySQL数据库连接 </a> </li> <li class=md-nav__item> <a href=#423-hbase class=md-nav__link> 4.2.3 HBase数据库 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#5-rdd class=md-nav__link> 第5章 RDD编程进阶 </a> </li> <li class=md-nav__item> <a href=#51 class=md-nav__link> 5.1 累加器 </a> <nav class=md-nav aria-label="5.1 累加器"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#511 class=md-nav__link> 5.1.1 系统累加器 </a> </li> <li class=md-nav__item> <a href=#512 class=md-nav__link> 5.1.2 自定义累加器 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#52 class=md-nav__link> 5.2 广播变量（调优策略） </a> </li> <li class=md-nav__item> <a href=#6 class=md-nav__link> 第6章 扩展 </a> </li> <li class=md-nav__item> <a href=#61-rdd class=md-nav__link> 6.1 RDD相关概念关系 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=../SparkSql/ title=SparkSql class=md-nav__link> SparkSql </a> </li> <li class=md-nav__item> <a href=../SparkStreaming/ title=SparkStreaming class=md-nav__link> SparkStreaming </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=../Flink/ title=Flink class=md-nav__link> Flink </a> </li> <li class=md-nav__item> <a href=../Zookeeper/ title=Zookeeper class=md-nav__link> Zookeeper </a> </li> <li class=md-nav__item> <a href=../Hbase/ title=HBase class=md-nav__link> HBase </a> </li> <li class=md-nav__item> <a href=../Hive/ title=Hive class=md-nav__link> Hive </a> </li> <li class=md-nav__item> <a href=../Kafka/ title=Kafka class=md-nav__link> Kafka </a> </li> <li class=md-nav__item> <a href=../Flume/ title=Flume class=md-nav__link> Flume </a> </li> <li class=md-nav__item> <a href=../Sqoop/ title=Sqoop class=md-nav__link> Sqoop </a> </li> <li class=md-nav__item> <a href=../Oozie/ title=Oozie class=md-nav__link> Oozie </a> </li> <li class=md-nav__item> <a href=../Azkaban/ title=Azkaban class=md-nav__link> Azkaban </a> </li> <li class=md-nav__item> <a href=../Kylin/ title=Kylin class=md-nav__link> Kylin </a> </li> <li class=md-nav__item> <a href=../Presto/ title=Presto class=md-nav__link> Presto </a> </li> <li class=md-nav__item> <a href=../ELK/ title=ELK class=md-nav__link> ELK </a> </li> </ul> </nav> </li> <li class="md-nav__item md-nav__item--nested"> <input class="md-nav__toggle md-toggle" data-md-toggle=nav-8 type=checkbox id=nav-8> <label class=md-nav__link for=nav-8> 数据库 <span class="md-nav__icon md-icon"></span> </label> <nav class=md-nav aria-label=数据库 data-md-level=1> <label class=md-nav__title for=nav-8> <span class="md-nav__icon md-icon"></span> 数据库 </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=../../%E5%BF%85%E5%A4%87%E6%8A%80%E6%9C%AF/Mysql%E5%AE%89%E8%A3%85/ title=Mysql安装 class=md-nav__link> Mysql安装 </a> </li> <li class=md-nav__item> <a href=../../%E5%BF%85%E5%A4%87%E6%8A%80%E6%9C%AF/Oracle%E5%AE%89%E8%A3%85/ title=Oracle安装 class=md-nav__link> Oracle安装 </a> </li> <li class=md-nav__item> <a href=../../%E5%BF%85%E5%A4%87%E6%8A%80%E6%9C%AF/Redis%E5%AE%89%E8%A3%85/ title=Redis安装 class=md-nav__link> Redis安装 </a> </li> </ul> </nav> </li> <li class="md-nav__item md-nav__item--nested"> <input class="md-nav__toggle md-toggle" data-md-toggle=nav-9 type=checkbox id=nav-9> <label class=md-nav__link for=nav-9> 项目开发 <span class="md-nav__icon md-icon"></span> </label> <nav class=md-nav aria-label=项目开发 data-md-level=1> <label class=md-nav__title for=nav-9> <span class="md-nav__icon md-icon"></span> 项目开发 </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=../../%E9%A1%B9%E7%9B%AE%E5%BC%80%E5%8F%91/Web%E5%BC%80%E5%8F%91/ title=Web开发 class=md-nav__link> Web开发 </a> </li> <li class=md-nav__item> <a href=../../%E9%A1%B9%E7%9B%AE%E5%BC%80%E5%8F%91/%E5%AE%89%E5%8D%93%E5%BC%80%E5%8F%91/ title=安卓开发 class=md-nav__link> 安卓开发 </a> </li> <li class=md-nav__item> <a href=../../%E9%A1%B9%E7%9B%AE%E5%BC%80%E5%8F%91/PC%E7%AB%AF%E5%BC%80%E5%8F%91/ title=PC端开发 class=md-nav__link> PC端开发 </a> </li> <li class=md-nav__item> <a href=../../%E9%A1%B9%E7%9B%AE%E5%BC%80%E5%8F%91/%E5%A4%A7%E6%95%B0%E6%8D%AE%E5%BC%80%E5%8F%91/ title=大数据开发 class=md-nav__link> 大数据开发 </a> </li> </ul> </nav> </li> <li class="md-nav__item md-nav__item--nested"> <input class="md-nav__toggle md-toggle" data-md-toggle=nav-10 type=checkbox id=nav-10> <label class=md-nav__link for=nav-10> 必备技能 <span class="md-nav__icon md-icon"></span> </label> <nav class=md-nav aria-label=必备技能 data-md-level=1> <label class=md-nav__title for=nav-10> <span class="md-nav__icon md-icon"></span> 必备技能 </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=../../%E5%BF%85%E5%A4%87%E6%8A%80%E6%9C%AF/Git/ title=Git class=md-nav__link> Git </a> </li> <li class=md-nav__item> <a href=../../%E5%BF%85%E5%A4%87%E6%8A%80%E6%9C%AF/Zabbix5/ title=Zabbix class=md-nav__link> Zabbix </a> </li> </ul> </nav> </li> <li class="md-nav__item md-nav__item--nested"> <input class="md-nav__toggle md-toggle" data-md-toggle=nav-11 type=checkbox id=nav-11> <label class=md-nav__link for=nav-11> 常用工具 <span class="md-nav__icon md-icon"></span> </label> <nav class=md-nav aria-label=常用工具 data-md-level=1> <label class=md-nav__title for=nav-11> <span class="md-nav__icon md-icon"></span> 常用工具 </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=../../%E5%B8%B8%E7%94%A8%E5%B7%A5%E5%85%B7/IDEA/ title=IDEA class=md-nav__link> IDEA </a> </li> <li class=md-nav__item> <a href=../../%E5%B8%B8%E7%94%A8%E5%B7%A5%E5%85%B7/Wiki/ title=Wiki class=md-nav__link> Wiki </a> </li> </ul> </nav> </li> <li class="md-nav__item md-nav__item--nested"> <input class="md-nav__toggle md-toggle" data-md-toggle=nav-12 type=checkbox id=nav-12> <label class=md-nav__link for=nav-12> 面试题库 <span class="md-nav__icon md-icon"></span> </label> <nav class=md-nav aria-label=面试题库 data-md-level=1> <label class=md-nav__title for=nav-12> <span class="md-nav__icon md-icon"></span> 面试题库 </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=../../%E5%B8%B8%E7%94%A8%E5%B7%A5%E5%85%B7/IDEA/ title=编程语言 class=md-nav__link> 编程语言 </a> </li> <li class=md-nav__item> <a href=../../%E5%B8%B8%E7%94%A8%E5%B7%A5%E5%85%B7/IDEA/ title=系统操作 class=md-nav__link> 系统操作 </a> </li> <li class=md-nav__item> <a href=../../%E5%B8%B8%E7%94%A8%E5%B7%A5%E5%85%B7/IDEA/ title=数据库 class=md-nav__link> 数据库 </a> </li> <li class=md-nav__item> <a href=../../%E5%B8%B8%E7%94%A8%E5%B7%A5%E5%85%B7/IDEA/ title=大数据 class=md-nav__link> 大数据 </a> </li> <li class=md-nav__item> <a href=../../%E5%B8%B8%E7%94%A8%E5%B7%A5%E5%85%B7/IDEA/ title=项目 class=md-nav__link> 项目 </a> </li> </ul> </nav> </li> </ul> </nav> </div> </div> </div> <div class="md-sidebar md-sidebar--secondary" data-md-component=toc> <div class=md-sidebar__scrollwrap> <div class=md-sidebar__inner> <nav class="md-nav md-nav--secondary" aria-label=目录> <label class=md-nav__title for=__toc> <span class="md-nav__icon md-icon"></span> 目录 </label> <ul class=md-nav__list data-md-scrollfix> <li class=md-nav__item> <a href=#1-rdd class=md-nav__link> 第1章 RDD概述 </a> </li> <li class=md-nav__item> <a href=#11-rdd class=md-nav__link> 1.1 什么是RDD </a> </li> <li class=md-nav__item> <a href=#12-rdd class=md-nav__link> 1.2 RDD的属性 </a> </li> <li class=md-nav__item> <a href=#13-rdd class=md-nav__link> 1.3 RDD特点 </a> <nav class=md-nav aria-label="1.3 RDD特点"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#131 class=md-nav__link> 1.3.1 分区 </a> </li> <li class=md-nav__item> <a href=#132 class=md-nav__link> 1.3.2 只读 </a> </li> <li class=md-nav__item> <a href=#133 class=md-nav__link> 1.3.3 依赖 </a> </li> <li class=md-nav__item> <a href=#134 class=md-nav__link> 1.3.4 缓存 </a> </li> <li class=md-nav__item> <a href=#135-checkpoint class=md-nav__link> 1.3.5 CheckPoint </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#2-rdd class=md-nav__link> 第2章 RDD编程 </a> </li> <li class=md-nav__item> <a href=#21 class=md-nav__link> 2.1 编程模型 </a> </li> <li class=md-nav__item> <a href=#22-rdd class=md-nav__link> 2.2 RDD的创建 </a> <nav class=md-nav aria-label="2.2 RDD的创建"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#221 class=md-nav__link> 2.2.1 从集合中创建 </a> </li> <li class=md-nav__item> <a href=#222 class=md-nav__link> 2.2.2 由外部存储系统的数据集创建 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#23-rdd class=md-nav__link> 2.3 RDD的转换（面试开发重点） </a> <nav class=md-nav aria-label="2.3 RDD的转换（面试开发重点）"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#231-value class=md-nav__link> 2.3.1 Value类型 </a> <nav class=md-nav aria-label="2.3.1 Value类型"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#2311-mapfunc class=md-nav__link> 2.3.1.1 map(func)案例 </a> </li> <li class=md-nav__item> <a href=#2312-mappartitionsfunc class=md-nav__link> 2.3.1.2 mapPartitions(func) 案例 </a> </li> <li class=md-nav__item> <a href=#2313-mappartitionswithindexfunc class=md-nav__link> 2.3.1.3 mapPartitionsWithIndex(func) 案例 </a> </li> <li class=md-nav__item> <a href=#2314-flatmapfunc class=md-nav__link> 2.3.1.4 flatMap(func) 案例 </a> </li> <li class=md-nav__item> <a href=#2315-mapmappartition class=md-nav__link> 2.3.1.5 map()和mapPartition()的区别 </a> </li> <li class=md-nav__item> <a href=#2316-glom class=md-nav__link> 2.3.1.6 glom案例 </a> </li> <li class=md-nav__item> <a href=#2317-groupbyfunc class=md-nav__link> 2.3.1.7 groupBy(func)案例 </a> </li> <li class=md-nav__item> <a href=#2318-filterfunc class=md-nav__link> 2.3.1.8 filter(func) 案例 </a> </li> <li class=md-nav__item> <a href=#2319-samplewithreplacement-fraction-seed class=md-nav__link> 2.3.1.9 sample(withReplacement, fraction, seed) 案例 </a> </li> <li class=md-nav__item> <a href=#23110-distinctnumtasks class=md-nav__link> 2.3.1.10 distinct([numTasks])) 案例 </a> </li> <li class=md-nav__item> <a href=#23111-coalescenumpartitions class=md-nav__link> 2.3.1.11 coalesce(numPartitions) 案例 </a> </li> <li class=md-nav__item> <a href=#23112-repartitionnumpartitions class=md-nav__link> 2.3.1.12 repartition(numPartitions) 案例 </a> </li> <li class=md-nav__item> <a href=#23113-coalescerepartition class=md-nav__link> 2.3.1.13 coalesce和repartition的区别 </a> </li> <li class=md-nav__item> <a href=#23114-sortbyfuncascending-numtasks class=md-nav__link> 2.3.1.14 sortBy(func,[ascending], [numTasks]) 案例 </a> </li> <li class=md-nav__item> <a href=#23115-pipecommand-envvars class=md-nav__link> 2.3.1.15 pipe(command, [envVars]) 案例 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#232-value class=md-nav__link> 2.3.2 双Value类型交互 </a> <nav class=md-nav aria-label="2.3.2 双Value类型交互"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#2321-unionotherdataset class=md-nav__link> 2.3.2.1 union(otherDataset) 案例 </a> </li> <li class=md-nav__item> <a href=#2322-subtract-otherdataset class=md-nav__link> 2.3.2.2 subtract (otherDataset) 案例 </a> </li> <li class=md-nav__item> <a href=#2323-intersectionotherdataset class=md-nav__link> 2.3.2.3 intersection(otherDataset) 案例 </a> </li> <li class=md-nav__item> <a href=#2324-cartesianotherdataset class=md-nav__link> 2.3.2.4 cartesian(otherDataset) 案例 </a> </li> <li class=md-nav__item> <a href=#2325-zipotherdataset class=md-nav__link> 2.3.2.5 zip(otherDataset)案例 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#233-key-value class=md-nav__link> 2.3.3 Key-Value类型 </a> <nav class=md-nav aria-label="2.3.3 Key-Value类型"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#2331-partitionby class=md-nav__link> 2.3.3.1 partitionBy案例 </a> </li> <li class=md-nav__item> <a href=#2332-groupbykey class=md-nav__link> 2.3.3.2 groupByKey案例 </a> </li> <li class=md-nav__item> <a href=#2333-reducebykeyfunc-numtasks class=md-nav__link> 2.3.3.3 reduceByKey(func, [numTasks]) 案例 </a> </li> <li class=md-nav__item> <a href=#2334-reducebykeygroupbykey class=md-nav__link> 2.3.3.4 reduceByKey和groupByKey的区别 </a> </li> <li class=md-nav__item> <a href=#2335-aggregatebykey class=md-nav__link> 2.3.3.5 aggregateByKey案例 </a> </li> <li class=md-nav__item> <a href=#2336-foldbykey class=md-nav__link> 2.3.3.6 foldByKey案例 </a> </li> <li class=md-nav__item> <a href=#2337-combinebykeyc class=md-nav__link> 2.3.3.7 combineByKey[C] 案例 </a> </li> <li class=md-nav__item> <a href=#2338-sortbykeyascending-numtasks class=md-nav__link> 2.3.3.8 sortByKey([ascending], [numTasks]) 案例 </a> </li> <li class=md-nav__item> <a href=#2339-mapvalues class=md-nav__link> 2.3.3.9 mapValues案例 </a> </li> <li class=md-nav__item> <a href=#23310-joinotherdataset-numtasks class=md-nav__link> 2.3.3.10 join(otherDataset, [numTasks]) 案例 </a> </li> <li class=md-nav__item> <a href=#23311-cogroupotherdataset-numtasks class=md-nav__link> 2.3.3.11 cogroup(otherDataset, [numTasks]) 案例 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#234 class=md-nav__link> 2.3.4 案例实操 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#24-action class=md-nav__link> 2.4 Action </a> <nav class=md-nav aria-label="2.4 Action"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#241-reducefunc class=md-nav__link> 2.4.1 reduce(func)案例 </a> </li> <li class=md-nav__item> <a href=#242-collect class=md-nav__link> 2.4.2 collect()案例 </a> </li> <li class=md-nav__item> <a href=#243-count class=md-nav__link> 2.4.3 count()案例 </a> </li> <li class=md-nav__item> <a href=#244-first class=md-nav__link> 2.4.4 first()案例 </a> </li> <li class=md-nav__item> <a href=#245-taken class=md-nav__link> 2.4.5 take(n)案例 </a> </li> <li class=md-nav__item> <a href=#246-takeorderedn class=md-nav__link> 2.4.6 takeOrdered(n)案例 </a> </li> <li class=md-nav__item> <a href=#247-aggregate class=md-nav__link> 2.4.7 aggregate案例 </a> </li> <li class=md-nav__item> <a href=#248-foldnumfunc class=md-nav__link> 2.4.8 fold(num)(func)案例 </a> </li> <li class=md-nav__item> <a href=#249-saveastextfilepath class=md-nav__link> 2.4.9 saveAsTextFile(path) </a> </li> <li class=md-nav__item> <a href=#2410-saveassequencefilepath class=md-nav__link> 2.4.10 saveAsSequenceFile(path) </a> </li> <li class=md-nav__item> <a href=#2411-saveasobjectfilepath class=md-nav__link> 2.4.11 saveAsObjectFile(path) </a> </li> <li class=md-nav__item> <a href=#2412-countbykey class=md-nav__link> 2.4.12 countByKey()案例 </a> </li> <li class=md-nav__item> <a href=#2413-foreachfunc class=md-nav__link> 2.4.13 foreach(func)案例 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#25-rdd class=md-nav__link> 2.5 RDD中的函数传递 </a> <nav class=md-nav aria-label="2.5 RDD中的函数传递"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#251 class=md-nav__link> 2.5.1 传递一个方法 </a> </li> <li class=md-nav__item> <a href=#252 class=md-nav__link> 2.5.2 传递一个属性 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#26-rdd class=md-nav__link> 2.6 RDD依赖关系 </a> <nav class=md-nav aria-label="2.6 RDD依赖关系"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#261-lineage class=md-nav__link> 2.6.1 Lineage </a> </li> <li class=md-nav__item> <a href=#262 class=md-nav__link> 2.6.2 窄依赖 </a> </li> <li class=md-nav__item> <a href=#263 class=md-nav__link> 2.6.3 宽依赖 </a> </li> <li class=md-nav__item> <a href=#264-dag class=md-nav__link> 2.6.4 DAG </a> </li> <li class=md-nav__item> <a href=#265 class=md-nav__link> 2.6.5 任务划分（面试重点） </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#27-rdd class=md-nav__link> 2.7 RDD缓存 </a> </li> <li class=md-nav__item> <a href=#28-rdd-checkpoint class=md-nav__link> 2.8 RDD CheckPoint </a> </li> <li class=md-nav__item> <a href=#3-rdd class=md-nav__link> 第3章 键值对RDD数据分区器 </a> </li> <li class=md-nav__item> <a href=#31-rdd class=md-nav__link> 3.1 获取RDD分区 </a> </li> <li class=md-nav__item> <a href=#32-hash class=md-nav__link> 3.2 Hash分区 </a> </li> <li class=md-nav__item> <a href=#33-ranger class=md-nav__link> 3.3 Ranger分区 </a> </li> <li class=md-nav__item> <a href=#34 class=md-nav__link> 3.4 自定义分区 </a> </li> <li class=md-nav__item> <a href=#4 class=md-nav__link> 第4章 数据读取与保存 </a> </li> <li class=md-nav__item> <a href=#_1 class=md-nav__link> 文件类数据读取与保存 </a> <nav class=md-nav aria-label=文件类数据读取与保存> <ul class=md-nav__list> <li class=md-nav__item> <a href=#411-text class=md-nav__link> 4.1.1 Text文件 </a> </li> <li class=md-nav__item> <a href=#412-json class=md-nav__link> 4.1.2 Json文件 </a> </li> <li class=md-nav__item> <a href=#413-sequence class=md-nav__link> 4.1.3 Sequence文件 </a> </li> <li class=md-nav__item> <a href=#414 class=md-nav__link> 4.1.4 对象文件 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#_2 class=md-nav__link> 文件系统类数据读取与保存 </a> <nav class=md-nav aria-label=文件系统类数据读取与保存> <ul class=md-nav__list> <li class=md-nav__item> <a href=#421-hdfs class=md-nav__link> 4.2.1 HDFS </a> </li> <li class=md-nav__item> <a href=#422-mysql class=md-nav__link> 4.2.2 MySQL数据库连接 </a> </li> <li class=md-nav__item> <a href=#423-hbase class=md-nav__link> 4.2.3 HBase数据库 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#5-rdd class=md-nav__link> 第5章 RDD编程进阶 </a> </li> <li class=md-nav__item> <a href=#51 class=md-nav__link> 5.1 累加器 </a> <nav class=md-nav aria-label="5.1 累加器"> <ul class=md-nav__list> <li class=md-nav__item> <a href=#511 class=md-nav__link> 5.1.1 系统累加器 </a> </li> <li class=md-nav__item> <a href=#512 class=md-nav__link> 5.1.2 自定义累加器 </a> </li> </ul> </nav> </li> <li class=md-nav__item> <a href=#52 class=md-nav__link> 5.2 广播变量（调优策略） </a> </li> <li class=md-nav__item> <a href=#6 class=md-nav__link> 第6章 扩展 </a> </li> <li class=md-nav__item> <a href=#61-rdd class=md-nav__link> 6.1 RDD相关概念关系 </a> </li> </ul> </nav> </div> </div> </div> <div class=md-content> <article class="md-content__inner md-typeset"> <h1 id=sparkcore><center>SparkCore</center></h1> <h2 id=1-rdd>第1章 RDD概述</h2> <h2 id=11-rdd>1.1 什么是RDD</h2> <p>RDD（Resilient Distributed Dataset）叫做分布式数据集，是Spark中最基本的数据抽象。代码中是一个抽象类，它代表一个不可变、可分区、里面的元素可并行计算的集合。</p> <h2 id=12-rdd>1.2 RDD的属性</h2> <p><img alt=QQ截图20151208094014 src=98f9781a601d0d2063bd520ed41e6048.png></p> <ol> <li> <p>一组分区（Partition），即数据集的基本组成单位;</p> </li> <li> <p>一个计算每个分区的函数;</p> </li> <li> <p>RDD之间的依赖关系;</p> </li> <li> <p>一个Partitioner，即RDD的分片函数;</p> </li> <li> <p>一个列表，存储存取每个Partition的优先位置（preferred location）。</p> </li> </ol> <h2 id=13-rdd>1.3 RDD特点</h2> <p>RDD表示只读的分区的数据集，对RDD进行改动，只能通过RDD的转换操作，由一个RDD得到一个新的RDD，新的RDD包含了从其他RDD衍生所必需的信息。RDDs之间存在依赖，RDD的执行是按照血缘关系延时计算的。如果血缘关系较长，可以通过持久化RDD来切断血缘关系。</p> <h3 id=131>1.3.1 分区</h3> <p>RDD逻辑上是分区的，每个分区的数据是抽象存在的，计算的时候会通过一个compute函数得到每个分区的数据。如果RDD是通过已有的文件系统构建，则compute函数是读取指定文件系统中的数据，如果RDD是通过其他RDD转换而来，则compute函数是执行转换逻辑将其他RDD的数据进行转换。</p> <p><img alt=dd-partition src=878a92e0b3435bc1ab04376f8498f0bb.png></p> <h3 id=132>1.3.2 只读</h3> <p>如下图所示，RDD是只读的，要想改变RDD中的数据，只能在现有的RDD基础上创建新的RDD。</p> <p><img alt=dd-readonly src=faa4705d63a4765cab0d88bd69da101f.png></p> <p>由一个RDD转换到另一个RDD，可以通过丰富的操作算子实现，不再像MapReduce那样只能写map和reduce了，如下图所示。</p> <p><img alt=dd-transform src=9d7bf932684c9a3797044014283377d5.png></p> <p>RDD的操作算子包括两类，一类叫做transformations，它是用来将RDD进行转化，构建RDD的血缘关系；另一类叫做actions，它是用来触发RDD的计算，得到RDD的相关计算结果或者将RDD保存的文件系统中。下图是RDD所支持的操作算子列表。</p> <h3 id=133>1.3.3 依赖</h3> <p>RDDs通过操作算子进行转换，转换得到的新RDD包含了从其他RDDs衍生所必需的信息，RDDs之间维护着这种血缘关系，也称之为依赖。如下图所示，依赖包括两种，一种是窄依赖，RDDs之间分区是一一对应的，另一种是宽依赖，下游RDD的每个分区与上游RDD(也称之为父RDD)的每个分区都有关，是多对多的关系。</p> <p><img alt=dd-dependency src=1bb62f3989fb34a114850e92d1c49d71.png></p> <h3 id=134>1.3.4 缓存</h3> <p>如果在应用程序中多次使用同一个RDD，可以将该RDD缓存起来，该RDD只有在第一次计算的时候会根据血缘关系得到分区的数据，在后续其他地方用到该RDD的时候，会直接从缓存处取而不用再根据血缘关系计算，这样就加速后期的重用。如下图所示，RDD-1经过一系列的转换后得到RDD-n并保存到hdfs，RDD-1在这一过程中会有个中间结果，如果将其缓存到内存，那么在随后的RDD-1转换到RDD-m这一过程中，就不会计算其之前的RDD-0了。</p> <p><img alt=dd-cache src=f962e59440536071e74f600ef154b664.png></p> <h3 id=135-checkpoint>1.3.5 CheckPoint</h3> <p>虽然RDD的血缘关系天然地可以实现容错，当RDD的某个分区数据失败或丢失，可以通过血缘关系重建。但是对于长时间迭代型应用来说，随着迭代的进行，RDDs之间的血缘关系会越来越长，一旦在后续迭代过程中出错，则需要通过非常长的血缘关系去重建，势必影响性能。为此，RDD支持checkpoint将数据保存到持久化的存储中，这样就可以切断之前的血缘关系，因为checkpoint后的RDD不需要知道它的父RDDs了，它可以从checkpoint处拿到数据。</p> <h2 id=2-rdd>第2章 RDD编程</h2> <h2 id=21>2.1 编程模型</h2> <p>在Spark中，RDD被表示为对象，通过对象上的方法调用来对RDD进行转换。经过一系列的transformations定义RDD之后，就可以调用actions触发RDD的计算，action可以是向应用程序返回结果(count, collect等)，或者是向存储系统保存数据(saveAsTextFile等)。在Spark中，只有遇到action，才会执行RDD的计算(即延迟计算)，这样在运行时可以通过管道的方式传输多个转换。</p> <p>要使用Spark，开发者需要编写一个Driver程序，它被提交到集群以调度运行Worker，如下图所示。Driver中定义了一个或多个RDD，并调用RDD上的action，Worker则执行RDD分区计算任务。</p> <p><img alt=park-runtime src=5a1a60df66a578e03f404b68c7c07abc.png></p> <p><img alt=../../../../../../Desktop/spark-structure. src=1e1cf852e6c8cb6faded81e690cc82f3.png></p> <h2 id=22-rdd>2.2 RDD的创建</h2> <p>在Spark中创建RDD的创建方式可以分为三种：从集合中创建RDD；从外部存储创建RDD；从其他RDD创建。</p> <h3 id=221>2.2.1 从集合中创建</h3> <p>从集合中创建RDD，<a href=https://www.iteblog.com/archives/tag/spark/ >Spark</a>主要提供了两种函数：parallelize和makeRDD</p> <p>1）使用parallelize()从集合创建</p> <p>scala> val rdd = sc.parallelize(Array(1,2,3,4,5,6,7,8))</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at \&lt;console>:24</p> <p>2）使用makeRDD()从集合创建</p> <p>scala> val rdd1 = sc.makeRDD(Array(1,2,3,4,5,6,7,8))</p> <p>rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at makeRDD at \&lt;console>:24</p> <table> <thead> <tr> <th></th> </tr> </thead> <tbody> <tr> <td></td> </tr> </tbody> </table> <h3 id=222>2.2.2 由外部存储系统的数据集创建</h3> <p>包括本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、Cassandra、HBase等，我们会在第4章详细介绍。</p> <p>scala> val rdd2= sc.textFile(&ldquo;hdfs://hadoop102:9000/RELEASE&rdquo;)</p> <p>rdd2: org.apache.spark.rdd.RDD[String] = hdfs:// hadoop102:9000/RELEASE MapPartitionsRDD[4] at textFile at \&lt;console>:24</p> <p>2.2.3 从其他RDD创建</p> <p>详见2.3节</p> <h2 id=23-rdd>2.3 RDD的转换（面试开发重点）</h2> <p>RDD整体上分为Value类型和Key-Value类型</p> <h3 id=231-value>2.3.1 Value类型</h3> <h4 id=2311-mapfunc>2.3.1.1 map(func)案例</h4> <ol> <li> <p>作用：返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成</p> </li> <li> <p>需求：创建一个1-10数组的RDD，将所有元素*2形成新的RDD</p> </li> </ol> <p>（1）创建</p> <p>scala> var source = sc.parallelize(1 to 10)</p> <p>source: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[8] at parallelize at \&lt;console>:24</p> <p>（2）打印</p> <p>scala> source.collect()</p> <p>res7: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)</p> <p>（3）将所有元素*2</p> <p>scala> val mapadd = source.map(_ * 2)</p> <p>mapadd: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[9] at map at \&lt;console>:26</p> <p>（4）打印最终结果</p> <p>scala> mapadd.collect()</p> <p>res8: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18, 20)</p> <h4 id=2312-mappartitionsfunc>2.3.1.2 mapPartitions(func) 案例</h4> <p>1. 作用：类似于map，但独立地在RDD的每一个分片上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T] => Iterator[U]。假设有N个元素，有M个分区，那么map的函数的将被调用N次,而mapPartitions被调用M次,一个函数一次处理所有分区。</p> <ol> <li>需求：创建一个RDD，使每个元素*2组成新的RDD</li> </ol> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(Array(1,2,3,4))</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at \&lt;console>:24</p> <p>（2）使每个元素*2组成新的RDD</p> <p>scala> rdd.mapPartitions(x=>x.map(_*2))</p> <p>res3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[6] at mapPartitions at \&lt;console>:27</p> <p>（3）打印新的RDD</p> <p>scala> res3.collect</p> <p>res4: Array[Int] = Array(2, 4, 6, 8)</p> <h4 id=2313-mappartitionswithindexfunc>2.3.1.3 mapPartitionsWithIndex(func) 案例</h4> <p>1. 作用：类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是(Int, Interator[T]) => Iterator[U]；</p> <ol> <li>需求：创建一个RDD，使每个元素跟所在分区形成一个元组组成一个新的RDD</li> </ol> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(Array(1,2,3,4))</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at \&lt;console>:24</p> <p>（2）使每个元素跟所在分区形成一个元组组成一个新的RDD</p> <p>scala> val indexRdd = rdd.mapPartitionsWithIndex((index,items)=>(items.map((index,_))))</p> <p>indexRdd: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[5] at mapPartitionsWithIndex at \&lt;console>:26</p> <p>（3）打印新的RDD</p> <p>scala> indexRdd.collect</p> <p>res2: Array[(Int, Int)] = Array((0,1), (0,2), (1,3), (1,4))</p> <h4 id=2314-flatmapfunc>2.3.1.4 flatMap(func) 案例</h4> <p>1. 作用：类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素）</p> <p>2. 需求：创建一个元素为1-5的RDD，运用flatMap创建一个新的RDD，新的RDD为原RDD的每个元素的2倍（2，4，6，8，10）</p> <p>（1）创建</p> <p>scala> val sourceFlat = sc.parallelize(1 to 5)</p> <p>sourceFlat: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[12] at parallelize at \&lt;console>:24</p> <p>（2）打印</p> <p>scala> sourceFlat.collect()</p> <p>res11: Array[Int] = Array(1, 2, 3, 4, 5)</p> <p>（3）根据原RDD创建新RDD（1->1,2->1,2……5->1,2,3,4,5）</p> <p>scala> val flatMap = sourceFlat.flatMap(1 to _)</p> <p>flatMap: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[13] at flatMap at \&lt;console>:26</p> <p>（4）打印新RDD</p> <p>scala> flatMap.collect()</p> <p>res12: Array[Int] = Array(1, 1, 2, 1, 2, 3, 1, 2, 3, 4, 1, 2, 3, 4, 5)</p> <h4 id=2315-mapmappartition>2.3.1.5 map()和mapPartition()的区别</h4> <ol> <li>map()：每次处理一条数据。</li> </ol> <p>2. mapPartition()：每次处理一个分区的数据，这个分区的数据处理完后，原RDD中分区的数据才能释放，可能导致OOM。</p> <ol> <li>开发指导：当内存空间较大的时候建议使用mapPartition()，以提高处理效率。</li> </ol> <h4 id=2316-glom>2.3.1.6 glom案例</h4> <ol> <li> <p>作用：将每一个分区形成一个数组，形成新的RDD类型时RDD[Array[T]]</p> </li> <li> <p>需求：创建一个4个分区的RDD，并将每个分区的数据放到一个数组</p> </li> </ol> <p>（1）创建</p> <p>scala> val rdd = sc.parallelize(1 to 16,4)</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[65] at parallelize at \&lt;console>:24</p> <p>（2）将每个分区的数据放到一个数组并收集到Driver端打印</p> <p>scala> rdd.glom().collect()</p> <p>res25: Array[Array[Int]] = Array(Array(1, 2, 3, 4), Array(5, 6, 7, 8), Array(9, 10, 11, 12), Array(13, 14, 15, 16))</p> <h4 id=2317-groupbyfunc>2.3.1.7 groupBy(func)案例</h4> <ol> <li> <p>作用：分组，按照传入函数的返回值进行分组。将相同的key对应的值放入一个迭代器。</p> </li> <li> <p>需求：创建一个RDD，按照元素模以2的值进行分组。</p> </li> </ol> <p>（1）创建</p> <p>scala> val rdd = sc.parallelize(1 to 4)</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[65] at parallelize at \&lt;console>:24</p> <p>（2）按照元素模以2的值进行分组</p> <p>scala> val group = rdd.groupBy(_%2)</p> <p>group: org.apache.spark.rdd.RDD[(Int, Iterable[Int])] = ShuffledRDD[2] at groupBy at \&lt;console>:26</p> <p>（3）打印结果</p> <p>scala> group.collect</p> <p>res0: Array[(Int, Iterable[Int])] = Array((0,CompactBuffer(2, 4)), (1,CompactBuffer(1, 3)))</p> <h4 id=2318-filterfunc>2.3.1.8 filter(func) 案例</h4> <p>1. 作用：过滤。返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成。</p> <ol> <li>需求：创建一个RDD（由字符串组成），过滤出一个新RDD（包含”xiao”子串）</li> </ol> <p>（1）创建</p> <p>scala> var sourceFilter = sc.parallelize(Array(&ldquo;xiaoming&rdquo;,&rdquo;xiaojiang&rdquo;,&rdquo;xiaohe&rdquo;,&rdquo;dazhi&rdquo;))</p> <p>sourceFilter: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[10] at parallelize at \&lt;console>:24</p> <p>（2）打印</p> <p>scala> sourceFilter.collect()</p> <p>res9: Array[String] = Array(xiaoming, xiaojiang, xiaohe, dazhi)</p> <p>（3）过滤出含” xiao”子串的形成一个新的RDD</p> <p>scala> val filter = sourceFilter.filter(_.contains(&ldquo;xiao&rdquo;))</p> <p>filter: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[11] at filter at \&lt;console>:26</p> <p>（4）打印新RDD</p> <p>scala> filter.collect()</p> <p>res10: Array[String] = Array(xiaoming, xiaojiang, xiaohe)</p> <h4 id=2319-samplewithreplacement-fraction-seed>2.3.1.9 sample(withReplacement, fraction, seed) 案例</h4> <p>1. 作用：以指定的随机种子随机抽样出数量为fraction的数据，withReplacement表示是抽出的数据是否放回，true为有放回的抽样，false为无放回的抽样，seed用于指定随机数生成器种子。</p> <ol> <li>需求：创建一个RDD（1-10），从中选择放回和不放回抽样</li> </ol> <p>（1）创建RDD</p> <p>scala> val rdd = sc.parallelize(1 to 10)</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[20] at parallelize at \&lt;console>:24</p> <p>（2）打印</p> <p>scala> rdd.collect()</p> <p>res15: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)</p> <p>（3）放回抽样</p> <p>scala> var sample1 = rdd.sample(true,0.4,2)</p> <p>sample1: org.apache.spark.rdd.RDD[Int] = PartitionwiseSampledRDD[21] at sample at \&lt;console>:26</p> <p>（4）打印放回抽样结果</p> <p>scala> sample1.collect()</p> <p>res16: Array[Int] = Array(1, 2, 2, 7, 7, 8, 9)</p> <p>（5）不放回抽样</p> <p>scala> var sample2 = rdd.sample(false,0.2,3)</p> <p>sample2: org.apache.spark.rdd.RDD[Int] = PartitionwiseSampledRDD[22] at sample at \&lt;console>:26</p> <p>（6）打印不放回抽样结果</p> <p>scala> sample2.collect()</p> <p>res17: Array[Int] = Array(1, 9)</p> <h4 id=23110-distinctnumtasks>2.3.1.10 distinct([numTasks])) 案例</h4> <p>1. 作用：对源RDD进行去重后返回一个新的RDD。默认情况下，只有8个并行任务来操作，但是可以传入一个可选的numTasks参数改变它。</p> <ol> <li>需求：创建一个RDD，使用distinct()对其去重。</li> </ol> <p>（1）创建一个RDD</p> <p>scala> val distinctRdd = sc.parallelize(List(1,2,1,5,2,9,6,1))</p> <p>distinctRdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[34] at parallelize at \&lt;console>:24</p> <p>（2）对RDD进行去重（不指定并行度）</p> <p>scala> val unionRDD = distinctRdd.distinct()</p> <p>unionRDD: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[37] at distinct at \&lt;console>:26</p> <p>（3）打印去重后生成的新RDD</p> <p>scala> unionRDD.collect()</p> <p>res20: Array[Int] = Array(1, 9, 5, 6, 2)</p> <p>（4）对RDD（指定并行度为2）</p> <p>scala> val unionRDD = distinctRdd.distinct(2)</p> <p>unionRDD: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[40] at distinct at \&lt;console>:26</p> <p>（5）打印去重后生成的新RDD</p> <p>scala> unionRDD.collect()</p> <p>res21: Array[Int] = Array(6, 2, 1, 9, 5)</p> <h4 id=23111-coalescenumpartitions>2.3.1.11 coalesce(numPartitions) 案例</h4> <ol> <li> <p>作用：缩减分区数，用于大数据集过滤后，提高小数据集的执行效率。</p> </li> <li> <p>需求：创建一个4个分区的RDD，对其缩减分区</p> </li> </ol> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(1 to 16,4)</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[54] at parallelize at \&lt;console>:24</p> <p>（2）查看RDD的分区数</p> <p>scala> rdd.partitions.size</p> <p>res20: Int = 4</p> <p>（3）对RDD重新分区</p> <p>scala> val coalesceRDD = rdd.coalesce(3)</p> <p>coalesceRDD: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[55] at coalesce at \&lt;console>:26</p> <p>（4）查看新RDD的分区数</p> <p>scala> coalesceRDD.partitions.size</p> <p>res21: Int = 3</p> <h4 id=23112-repartitionnumpartitions>2.3.1.12 repartition(numPartitions) 案例</h4> <ol> <li> <p>作用：根据分区数，重新通过网络随机洗牌所有数据。</p> </li> <li> <p>需求：创建一个4个分区的RDD，对其重新分区</p> </li> </ol> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(1 to 16,4)</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[56] at parallelize at \&lt;console>:24</p> <p>（2）查看RDD的分区数</p> <p>scala> rdd.partitions.size</p> <p>res22: Int = 4</p> <p>（3）对RDD重新分区</p> <p>scala> val rerdd = rdd.repartition(2)</p> <p>rerdd: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[60] at repartition at \&lt;console>:26</p> <p>（4）查看新RDD的分区数</p> <p>scala> rerdd.partitions.size</p> <p>res23: Int = 2</p> <h4 id=23113-coalescerepartition>2.3.1.13 coalesce和repartition的区别</h4> <ol> <li> <p>coalesce重新分区，可以选择是否进行shuffle过程。由参数shuffle: Boolean = false/true决定。</p> </li> <li> <p>repartition实际上是调用的coalesce，默认是进行shuffle的。源码如下：</p> </li> </ol> <p>def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {<br> coalesce(numPartitions, shuffle = true)<br> }</p> <h4 id=23114-sortbyfuncascending-numtasks>2.3.1.14 sortBy(func,[ascending], [numTasks]) 案例</h4> <ol> <li> <p>作用；使用func先对数据进行处理，按照处理后的数据比较结果排序，默认为正序。</p> </li> <li> <p>需求：创建一个RDD，按照不同的规则进行排序</p> </li> </ol> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(List(2,1,3,4))</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[21] at parallelize at \&lt;console>:24</p> <p>（2）按照自身大小排序</p> <p>scala> rdd.sortBy(x => x).collect()</p> <p>res11: Array[Int] = Array(1, 2, 3, 4)</p> <p>（3）按照与3余数的大小排序</p> <p>scala> rdd.sortBy(x => x%3).collect()</p> <p>res12: Array[Int] = Array(3, 4, 1, 2)</p> <h4 id=23115-pipecommand-envvars>2.3.1.15 pipe(command, [envVars]) 案例</h4> <ol> <li>作用：管道，针对每个分区，都执行一个shell脚本，返回输出的RDD。</li> </ol> <p>注意：脚本需要放在Worker节点可以访问到的位置</p> <ol> <li>需求：编写一个脚本，使用管道将脚本作用于RDD上。</li> </ol> <p>（1）编写一个脚本</p> <p>Shell脚本</p> <p>#!/bin/sh</p> <p>echo &ldquo;AA&rdquo;</p> <p>while read LINE; do</p> <p>echo &ldquo;>>>&ldquo;${LINE}</p> <p>done</p> <p>（2）创建一个只有一个分区的RDD</p> <p>scala> val rdd = sc.parallelize(List(&ldquo;hi&rdquo;,&rdquo;Hello&rdquo;,&rdquo;how&rdquo;,&rdquo;are&rdquo;,&rdquo;you&rdquo;),1)</p> <p>rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[50] at parallelize at \&lt;console>:24</p> <p>（3）将脚本作用该RDD并打印</p> <p>scala> rdd.pipe(&ldquo;/opt/module/spark/pipe.sh&rdquo;).collect()</p> <p>res18: Array[String] = Array(AA, >>>hi, >>>Hello, >>>how, >>>are, >>>you)</p> <p>（4）创建一个有两个分区的RDD</p> <p>scala> val rdd = sc.parallelize(List(&ldquo;hi&rdquo;,&rdquo;Hello&rdquo;,&rdquo;how&rdquo;,&rdquo;are&rdquo;,&rdquo;you&rdquo;),2)</p> <p>rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[52] at parallelize at \&lt;console>:24</p> <p>（5）将脚本作用该RDD并打印</p> <p>scala> rdd.pipe(&ldquo;/opt/module/spark/pipe.sh&rdquo;).collect()</p> <p>res19: Array[String] = Array(AA, >>>hi, >>>Hello, AA, >>>how, >>>are, >>>you)</p> <h3 id=232-value>2.3.2 双Value类型交互</h3> <h4 id=2321-unionotherdataset>2.3.2.1 union(otherDataset) 案例</h4> <ol> <li> <p>作用：对源RDD和参数RDD求并集后返回一个新的RDD</p> </li> <li> <p>需求：创建两个RDD，求并集</p> </li> </ol> <p>（1）创建第一个RDD</p> <p>scala> val rdd1 = sc.parallelize(1 to 5)</p> <p>rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[23] at parallelize at \&lt;console>:24</p> <p>（2）创建第二个RDD</p> <p>scala> val rdd2 = sc.parallelize(5 to 10)</p> <p>rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[24] at parallelize at \&lt;console>:24</p> <p>（3）计算两个RDD的并集</p> <p>scala> val rdd3 = rdd1.union(rdd2)</p> <p>rdd3: org.apache.spark.rdd.RDD[Int] = UnionRDD[25] at union at \&lt;console>:28</p> <p>（4）打印并集结果</p> <p>scala> rdd3.collect()</p> <p>res18: Array[Int] = Array(1, 2, 3, 4, 5, 5, 6, 7, 8, 9, 10)</p> <h4 id=2322-subtract-otherdataset>2.3.2.2 subtract (otherDataset) 案例</h4> <ol> <li> <p>作用：计算差的一种函数，去除两个RDD中相同的元素，不同的RDD将保留下来</p> </li> <li> <p>需求：创建两个RDD，求第一个RDD与第二个RDD的差集</p> </li> </ol> <p>（1）创建第一个RDD</p> <p>scala> val rdd = sc.parallelize(3 to 8)</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[70] at parallelize at \&lt;console>:24</p> <p>（2）创建第二个RDD</p> <p>scala> val rdd1 = sc.parallelize(1 to 5)</p> <p>rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[71] at parallelize at \&lt;console>:24</p> <p>（3）计算第一个RDD与第二个RDD的差集并打印</p> <p>scala> rdd.subtract(rdd1).collect()</p> <p>res27: Array[Int] = Array(8, 6, 7)</p> <h4 id=2323-intersectionotherdataset>2.3.2.3 intersection(otherDataset) 案例</h4> <ol> <li> <p>作用：对源RDD和参数RDD求交集后返回一个新的RDD</p> </li> <li> <p>需求：创建两个RDD，求两个RDD的交集</p> </li> </ol> <p>（1）创建第一个RDD</p> <p>scala> val rdd1 = sc.parallelize(1 to 7)</p> <p>rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[26] at parallelize at \&lt;console>:24</p> <p>（2）创建第二个RDD</p> <p>scala> val rdd2 = sc.parallelize(5 to 10)</p> <p>rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[27] at parallelize at \&lt;console>:24</p> <p>（3）计算两个RDD的交集</p> <p>scala> val rdd3 = rdd1.intersection(rdd2)</p> <p>rdd3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[33] at intersection at \&lt;console>:28</p> <p>（4）打印计算结果</p> <p>scala> rdd3.collect()</p> <p>res19: Array[Int] = Array(5, 6, 7)</p> <h4 id=2324-cartesianotherdataset>2.3.2.4 cartesian(otherDataset) 案例</h4> <ol> <li> <p>作用：笛卡尔积（尽量避免使用）</p> </li> <li> <p>需求：创建两个RDD，计算两个RDD的笛卡尔积</p> </li> </ol> <p>（1）创建第一个RDD</p> <p>scala> val rdd1 = sc.parallelize(1 to 3)</p> <p>rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[47] at parallelize at \&lt;console>:24</p> <p>（2）创建第二个RDD</p> <p>scala> val rdd2 = sc.parallelize(2 to 5)</p> <p>rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[48] at parallelize at \&lt;console>:24</p> <p>（3）计算两个RDD的笛卡尔积并打印</p> <p>scala> rdd1.cartesian(rdd2).collect()</p> <p>res17: Array[(Int, Int)] = Array((1,2), (1,3), (1,4), (1,5), (2,2), (2,3), (2,4), (2,5), (3,2), (3,3), (3,4), (3,5))</p> <h4 id=2325-zipotherdataset>2.3.2.5 zip(otherDataset)案例</h4> <p>1. 作用：将两个RDD组合成Key/Value形式的RDD,这里默认两个RDD的partition数量以及元素数量都相同，否则会抛出异常。</p> <ol> <li>需求：创建两个RDD，并将两个RDD组合到一起形成一个(k,v)RDD</li> </ol> <p>（1）创建第一个RDD</p> <p>scala> val rdd1 = sc.parallelize(Array(1,2,3),3)</p> <p>rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at \&lt;console>:24</p> <p>（2）创建第二个RDD（与1分区数相同）</p> <p>scala> val rdd2 = sc.parallelize(Array(&ldquo;a&rdquo;,&rdquo;b&rdquo;,&rdquo;c&rdquo;),3)</p> <p>rdd2: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[2] at parallelize at \&lt;console>:24</p> <p>（3）第一个RDD组合第二个RDD并打印</p> <p>scala> rdd1.zip(rdd2).collect</p> <p>res1: Array[(Int, String)] = Array((1,a), (2,b), (3,c))</p> <p>（4）第二个RDD组合第一个RDD并打印</p> <p>scala> rdd2.zip(rdd1).collect</p> <p>res2: Array[(String, Int)] = Array((a,1), (b,2), (c,3))</p> <p>（5）创建第三个RDD（与1,2分区数不同）</p> <p>scala> val rdd3 = sc.parallelize(Array(&ldquo;a&rdquo;,&rdquo;b&rdquo;,&rdquo;c&rdquo;),2)</p> <p>rdd3: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[5] at parallelize at \&lt;console>:24</p> <p>（6）第一个RDD组合第三个RDD并打印</p> <p>scala> rdd1.zip(rdd3).collect</p> <p>java.lang.IllegalArgumentException: Can&rsquo;t zip RDDs with unequal numbers of partitions: List(3, 2)</p> <p>at org.apache.spark.rdd.ZippedPartitionsBaseRDD.getPartitions(ZippedPartitionsRDD.scala:57)</p> <p>at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:252)</p> <p>at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:250)</p> <p>at scala.Option.getOrElse(Option.scala:121)</p> <p>at org.apache.spark.rdd.RDD.partitions(RDD.scala:250)</p> <p>at org.apache.spark.SparkContext.runJob(SparkContext.scala:1965)</p> <p>at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:936)</p> <p>at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)</p> <p>at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)</p> <p>at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)</p> <p>at org.apache.spark.rdd.RDD.collect(RDD.scala:935)</p> <p>&hellip; 48 elided</p> <h3 id=233-key-value>2.3.3 Key-Value类型</h3> <h4 id=2331-partitionby>2.3.3.1 partitionBy案例</h4> <p>1. 作用：对pairRDD进行分区操作，如果原有的partionRDD和现有的partionRDD是一致的话就不进行分区， 否则会生成ShuffleRDD，即会产生shuffle过程。</p> <ol> <li>需求：创建一个4个分区的RDD，对其重新分区</li> </ol> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(Array((1,&rdquo;aaa&rdquo;),(2,&rdquo;bbb&rdquo;),(3,&rdquo;ccc&rdquo;),(4,&rdquo;ddd&rdquo;)),4)</p> <p>rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[44] at parallelize at \&lt;console>:24</p> <p>（2）查看RDD的分区数</p> <p>scala> rdd.partitions.size</p> <p>res24: Int = 4</p> <p>（3）对RDD重新分区</p> <p>scala> var rdd2 = rdd.partitionBy(new org.apache.spark.HashPartitioner(2))</p> <p>rdd2: org.apache.spark.rdd.RDD[(Int, String)] = ShuffledRDD[45] at partitionBy at \&lt;console>:26</p> <p>（4）查看新RDD的分区数</p> <p>scala> rdd2.partitions.size</p> <p>res25: Int = 2</p> <h4 id=2332-groupbykey>2.3.3.2 groupByKey案例</h4> <ol> <li>作用：groupByKey也是对每个key进行操作，但只生成一个sequence。</li> </ol> <p>2. 需求：创建一个pairRDD，将相同key对应值聚合到一个sequence中，并计算相同key对应值的相加结果。</p> <p>（1）创建一个pairRDD</p> <p>scala> val words = Array(&ldquo;one&rdquo;, &ldquo;two&rdquo;, &ldquo;two&rdquo;, &ldquo;three&rdquo;, &ldquo;three&rdquo;, &ldquo;three&rdquo;)</p> <p>words: Array[String] = Array(one, two, two, three, three, three)</p> <p>scala> val wordPairsRDD = sc.parallelize(words).map(word => (word, 1))</p> <p>wordPairsRDD: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[4] at map at \&lt;console>:26</p> <p>（2）将相同key对应值聚合到一个sequence中</p> <p>scala> val group = wordPairsRDD.groupByKey()</p> <p>group: org.apache.spark.rdd.RDD[(String, Iterable[Int])] = ShuffledRDD[5] at groupByKey at \&lt;console>:28</p> <p>（3）打印结果</p> <p>scala> group.collect()</p> <p>res1: Array[(String, Iterable[Int])] = Array((two,CompactBuffer(1, 1)), (one,CompactBuffer(1)), (three,CompactBuffer(1, 1, 1)))</p> <p>（4）计算相同key对应值的相加结果</p> <p>scala> group.map(t => (t._1, t._2.sum))</p> <p>res2: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[6] at map at \&lt;console>:31</p> <p>（5）打印结果</p> <p>scala> res2.collect()</p> <p>res3: Array[(String, Int)] = Array((two,2), (one,1), (three,3))</p> <h4 id=2333-reducebykeyfunc-numtasks>2.3.3.3 reduceByKey(func, [numTasks]) 案例</h4> <p>1. 在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，reduce任务的个数可以通过第二个可选的参数来设置。</p> <ol> <li>需求：创建一个pairRDD，计算相同key对应值的相加结果</li> </ol> <p>（1）创建一个pairRDD</p> <p>scala> val rdd = sc.parallelize(List((&ldquo;female&rdquo;,1),(&ldquo;male&rdquo;,5),(&ldquo;female&rdquo;,5),(&ldquo;male&rdquo;,2)))</p> <p>rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[46] at parallelize at \&lt;console>:24</p> <p>（2）计算相同key对应值的相加结果</p> <p>scala> val reduce = rdd.reduceByKey((x,y) => x+y)</p> <p>reduce: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[47] at reduceByKey at \&lt;console>:26</p> <p>（3）打印结果</p> <p>scala> reduce.collect()</p> <p>res29: Array[(String, Int)] = Array((female,6), (male,7))</p> <h4 id=2334-reducebykeygroupbykey>2.3.3.4 reduceByKey和groupByKey的区别</h4> <p>1. reduceByKey：按照key进行聚合，在shuffle之前有combine（预聚合）操作，返回结果是RDD[k,v].</p> <ol> <li> <p>groupByKey：按照key进行分组，直接进行shuffle。</p> </li> <li> <p>开发指导：reduceByKey比groupByKey，建议使用。但是需要注意是否会影响业务逻辑。</p> </li> </ol> <h4 id=2335-aggregatebykey>2.3.3.5 aggregateByKey案例</h4> <p>参数：(zeroValue:U,[partitioner: Partitioner]) (seqOp: (U, V) => U,combOp: (U, U) => U)</p> <p>1. 作用：在kv对的RDD中，，按key将value进行分组合并，合并时，将每个value和初始值作为seq函数的参数，进行计算，返回的结果作为一个新的kv对，然后再将结果按照key进行合并，最后将每个分组的value传递给combine函数进行计算（先将前两个value进行计算，将返回结果和下一个value传给combine函数，以此类推），将key与计算结果作为一个新的kv对输出。</p> <ol> <li>参数描述：</li> </ol> <p>（1）zeroValue：给每一个分区中的每一个key一个初始值；</p> <p>（2）seqOp：函数用于在每一个分区中用初始值逐步迭代value；</p> <p>（3）combOp：函数用于合并每个分区中的结果。</p> <ol> <li> <p>需求：创建一个pairRDD，取出每个分区相同key对应值的最大值，然后相加</p> </li> <li> <p>需求分析</p> </li> </ol> <p>图1-aggregate案例分析</p> <p>（1）创建一个pairRDD</p> <p>scala> val rdd = sc.parallelize(List((&ldquo;a&rdquo;,3),(&ldquo;a&rdquo;,2),(&ldquo;c&rdquo;,4),(&ldquo;b&rdquo;,3),(&ldquo;c&rdquo;,6),(&ldquo;c&rdquo;,8)),2)</p> <p>rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[0] at parallelize at \&lt;console>:24</p> <p>（2）取出每个分区相同key对应值的最大值，然后相加</p> <p>scala> val agg = rdd.aggregateByKey(0)(math.max(<em>,</em>),<em>+</em>)</p> <p>agg: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[1] at aggregateByKey at \&lt;console>:26</p> <p>（3）打印结果</p> <p>scala> agg.collect()</p> <p>res0: Array[(String, Int)] = Array((b,3), (a,3), (c,12))</p> <h4 id=2336-foldbykey>2.3.3.6 foldByKey案例</h4> <p>参数：(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]</p> <ol> <li> <p>作用：aggregateByKey的简化操作，seqop和combop相同</p> </li> <li> <p>需求：创建一个pairRDD，计算相同key对应值的相加结果</p> </li> </ol> <p>（1）创建一个pairRDD</p> <p>scala> val rdd = sc.parallelize(List((1,3),(1,2),(1,4),(2,3),(3,6),(3,8)),3)</p> <p>rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[91] at parallelize at \&lt;console>:24</p> <p>（2）计算相同key对应值的相加结果</p> <p>scala> val agg = rdd.foldByKey(0)(<em>+</em>)</p> <p>agg: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[92] at foldByKey at \&lt;console>:26</p> <p>（3）打印结果</p> <p>scala> agg.collect()</p> <p>res61: Array[(Int, Int)] = Array((3,14), (1,9), (2,3))</p> <h4 id=2337-combinebykeyc>2.3.3.7 combineByKey[C] 案例</h4> <p>参数：(createCombiner: V => C,  mergeValue: (C, V) => C,  mergeCombiners: (C, C) => C)</p> <ol> <li> <p>作用：对相同K，把V合并成一个集合。</p> </li> <li> <p>参数描述：</p> </li> </ol> <p>（1）createCombiner: combineByKey() 会遍历分区中的所有元素，因此每个元素的键要么还没有遇到过，要么就和之前的某个元素的键相同。如果这是一个新的元素,combineByKey()会使用一个叫作createCombiner()的函数来创建那个键对应的累加器的初始值</p> <p>（2）mergeValue: 如果这是一个在处理当前分区之前已经遇到的键，它会使用mergeValue()方法将该键的累加器对应的当前值与这个新的值进行合并</p> <p>（3）mergeCombiners: 由于每个分区都是独立处理的， 因此对于同一个键可以有多个累加器。如果有两个或者更多的分区都有对应同一个键的累加器， 就需要使用用户提供的 mergeCombiners() 方法将各个分区的结果进行合并。</p> <ol> <li> <p>需求：创建一个pairRDD，根据key计算每种key的均值。（先计算每个key出现的次数以及可以对应值的总和，再相除得到结果）</p> </li> <li> <p>需求分析：</p> </li> </ol> <p>图2- combineByKey案例分析</p> <p>（1）创建一个pairRDD</p> <p>scala> val input = sc.parallelize(Array((&ldquo;a&rdquo;, 88), (&ldquo;b&rdquo;, 95), (&ldquo;a&rdquo;, 91), (&ldquo;b&rdquo;, 93), (&ldquo;a&rdquo;, 95), (&ldquo;b&rdquo;, 98)),2)</p> <p>input: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[52] at parallelize at \&lt;console>:26</p> <p>（2）将相同key对应的值相加，同时记录该key出现的次数，放入一个二元组</p> <p>scala> val combine = input.combineByKey((_,1),(acc:(Int,Int),v)=>(acc._1+v,acc._2+1),(acc1:(Int,Int),acc2:(Int,Int))=>(acc1._1+acc2._1,acc1._2+acc2._2))</p> <p>combine: org.apache.spark.rdd.RDD[(String, (Int, Int))] = ShuffledRDD[5] at combineByKey at \&lt;console>:28</p> <p>（3）打印合并后的结果</p> <p>scala> combine.collect</p> <p>res5: Array[(String, (Int, Int))] = Array((b,(286,3)), (a,(274,3)))</p> <p>（4）计算平均值</p> <p>scala> val result = combine.map{case (key,value) => (key,value._1/value._2.toDouble)}</p> <p>result: org.apache.spark.rdd.RDD[(String, Double)] = MapPartitionsRDD[54] at map at \&lt;console>:30</p> <p>（5）打印结果</p> <p>scala> result.collect()</p> <p>res33: Array[(String, Double)] = Array((b,95.33333333333333), (a,91.33333333333333))</p> <h4 id=2338-sortbykeyascending-numtasks>2.3.3.8 sortByKey([ascending], [numTasks]) 案例</h4> <p>1. 作用：在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD</p> <ol> <li>需求：创建一个pairRDD，按照key的正序和倒序进行排序</li> </ol> <p>（1）创建一个pairRDD</p> <p>scala> val rdd = sc.parallelize(Array((3,&rdquo;aa&rdquo;),(6,&rdquo;cc&rdquo;),(2,&rdquo;bb&rdquo;),(1,&rdquo;dd&rdquo;)))</p> <p>rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[14] at parallelize at \&lt;console>:24</p> <p>（2）按照key的正序</p> <p>scala> rdd.sortByKey(true).collect()</p> <p>res9: Array[(Int, String)] = Array((1,dd), (2,bb), (3,aa), (6,cc))</p> <p>（3）按照key的倒序</p> <p>scala> rdd.sortByKey(false).collect()</p> <p>res10: Array[(Int, String)] = Array((6,cc), (3,aa), (2,bb), (1,dd))</p> <h4 id=2339-mapvalues>2.3.3.9 mapValues案例</h4> <ol> <li> <p>针对于(K,V)形式的类型只对V进行操作</p> </li> <li> <p>需求：创建一个pairRDD，并将value添加字符串&rdquo;|||&ldquo;</p> </li> </ol> <p>（1）创建一个pairRDD</p> <p>scala> val rdd3 = sc.parallelize(Array((1,&rdquo;a&rdquo;),(1,&rdquo;d&rdquo;),(2,&rdquo;b&rdquo;),(3,&rdquo;c&rdquo;)))</p> <p>rdd3: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[67] at parallelize at \&lt;console>:24</p> <p>（2）对value添加字符串&rdquo;|||&ldquo;</p> <p>scala> rdd3.mapValues(_+&rdquo;|||&ldquo;).collect()</p> <p>res26: Array[(Int, String)] = Array((1,a|||), (1,d|||), (2,b|||), (3,c|||))</p> <h4 id=23310-joinotherdataset-numtasks>2.3.3.10 join(otherDataset, [numTasks]) 案例</h4> <p>1. 作用：在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD</p> <ol> <li>需求：创建两个pairRDD，并将key相同的数据聚合到一个元组。</li> </ol> <p>（1）创建第一个pairRDD</p> <p>scala> val rdd = sc.parallelize(Array((1,&rdquo;a&rdquo;),(2,&rdquo;b&rdquo;),(3,&rdquo;c&rdquo;)))</p> <p>rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[32] at parallelize at \&lt;console>:24</p> <p>（2）创建第二个pairRDD</p> <p>scala> val rdd1 = sc.parallelize(Array((1,4),(2,5),(3,6)))</p> <p>rdd1: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[33] at parallelize at \&lt;console>:24</p> <p>（3）join操作并打印结果</p> <p>scala> rdd.join(rdd1).collect()</p> <p>res13: Array[(Int, (String, Int))] = Array((1,(a,4)), (2,(b,5)), (3,(c,6)))</p> <h4 id=23311-cogroupotherdataset-numtasks>2.3.3.11 cogroup(otherDataset, [numTasks]) 案例</h4> <p>1. 作用：在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable\&lt;V>,Iterable\&lt;W>))类型的RDD</p> <ol> <li>需求：创建两个pairRDD，并将key相同的数据聚合到一个迭代器。</li> </ol> <p>（1）创建第一个pairRDD</p> <p>scala> val rdd = sc.parallelize(Array((1,&rdquo;a&rdquo;),(2,&rdquo;b&rdquo;),(3,&rdquo;c&rdquo;)))</p> <p>rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[37] at parallelize at \&lt;console>:24</p> <p>（2）创建第二个pairRDD</p> <p>scala> val rdd1 = sc.parallelize(Array((1,4),(2,5),(3,6)))</p> <p>rdd1: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[38] at parallelize at \&lt;console>:24</p> <p>（3）cogroup两个RDD并打印结果</p> <p>scala> rdd.cogroup(rdd1).collect()</p> <p>res14: Array[(Int, (Iterable[String], Iterable[Int]))] = Array((1,(CompactBuffer(a),CompactBuffer(4))), (2,(CompactBuffer(b),CompactBuffer(5))), (3,(CompactBuffer&copy;,CompactBuffer(6))))</p> <h3 id=234>2.3.4 案例实操</h3> <ol> <li>数据结构：时间戳，省份，城市，用户，广告，中间字段使用空格分割。</li> </ol> <p>样本如下：</p> <p>1516609143867 6 7 64 16</p> <p>1516609143869 9 4 75 18</p> <p>1516609143869 1 7 87 12</p> <ol> <li> <p>需求：统计出每一个省份广告被点击次数的TOP3</p> </li> <li> <p>实现过程：</p> </li> </ol> <p>package com.package.practice</p> <p>import org.apache.spark.rdd.RDD</p> <p>import org.apache.spark.{SparkConf, SparkContext}</p> <p>//需求：统计出每一个省份广告被点击次数的TOP3</p> <p>object Practice {</p> <p>def main(args: Array[String]): Unit = {</p> <p>//1.初始化spark配置信息并建立与spark的连接</p> <p>val sparkConf = new SparkConf().setMaster(&ldquo;local[*]&rdquo;).setAppName(&ldquo;Practice&rdquo;)</p> <p>val sc = new SparkContext(sparkConf)</p> <p>//2.读取数据生成RDD：TS，Province，City，User，AD</p> <p>val line = sc.textFile(&ldquo;E:\\IDEAWorkSpace\\SparkTest\\src\\main\\resources\\agent.log&rdquo;)</p> <p>//3.按照最小粒度聚合：((Province,AD),1)</p> <p>val provinceAdToOne = line.map { x =></p> <p>val fields: Array[String] = x.split(&rdquo; &ldquo;)</p> <p>((fields(1), fields(4)), 1)</p> <p>}</p> <p>//4.计算每个省中每个广告被点击的总数：((Province,AD),sum)</p> <p>val provinceAdToSum = provinceAdToOne.reduceByKey(_ + _)</p> <p>//5.将省份作为key，广告加点击数为value：(Province,(AD,sum))</p> <p>val provinceToAdSum = provinceAdToSum.map(x => (x._1._1, (x._1._2, x._2)))</p> <p>//6.将同一个省份的所有广告进行聚合(Province,List((AD1,sum1),(AD2,sum2)&hellip;))</p> <p>val provinceGroup = provinceToAdSum.groupByKey()</p> <p>//7.对同一个省份所有广告的集合进行排序并取前3条，排序规则为广告点击总数</p> <p>val provinceAdTop3 = provinceGroup.mapValues { x =></p> <p>x.toList.sortWith((x, y) => x._2 > y._2).take(3)</p> <p>}</p> <p>//8.将数据拉取到Driver端并打印</p> <p>provinceAdTop3.collect().foreach(println)</p> <p>//9.关闭与spark的连接</p> <p>sc.stop()</p> <p>}</p> <p>}</p> <h2 id=24-action>2.4 Action</h2> <h3 id=241-reducefunc>2.4.1 reduce(func)案例</h3> <ol> <li> <p>作用：通过func函数聚集RDD中的所有元素，先聚合分区内数据，再聚合分区间数据。</p> </li> <li> <p>需求：创建一个RDD，将所有元素聚合得到结果。</p> </li> </ol> <p>（1）创建一个RDD[Int]</p> <p>scala> val rdd1 = sc.makeRDD(1 to 10,2)</p> <p>rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[85] at makeRDD at \&lt;console>:24</p> <p>（2）聚合RDD[Int]所有元素</p> <p>scala> rdd1.reduce(<em>+</em>)</p> <p>res50: Int = 55</p> <p>（3）创建一个RDD[String]</p> <p>scala> val rdd2 = sc.makeRDD(Array((&ldquo;a&rdquo;,1),(&ldquo;a&rdquo;,3),(&ldquo;c&rdquo;,3),(&ldquo;d&rdquo;,5)))</p> <p>rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[86] at makeRDD at \&lt;console>:24</p> <p>（4）聚合RDD[String]所有数据</p> <p>scala> rdd2.reduce((x,y)=>(x._1 + y._1,x._2 + y._2))</p> <p>res51: (String, Int) = (adca,12)</p> <h3 id=242-collect>2.4.2 collect()案例</h3> <ol> <li> <p>作用：在驱动程序中，以数组的形式返回数据集的所有元素。</p> </li> <li> <p>需求：创建一个RDD，并将RDD内容收集到Driver端打印</p> </li> </ol> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(1 to 10)</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at \&lt;console>:24</p> <p>（2）将结果收集到Driver端</p> <p>scala> rdd.collect</p> <p>res0: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)</p> <h3 id=243-count>2.4.3 count()案例</h3> <ol> <li> <p>作用：返回RDD中元素的个数</p> </li> <li> <p>需求：创建一个RDD，统计该RDD的条数</p> </li> </ol> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(1 to 10)</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at \&lt;console>:24</p> <p>（2）统计该RDD的条数</p> <p>scala> rdd.count</p> <p>res1: Long = 10</p> <h3 id=244-first>2.4.4 first()案例</h3> <ol> <li> <p>作用：返回RDD中的第一个元素</p> </li> <li> <p>需求：创建一个RDD，返回该RDD中的第一个元素</p> </li> </ol> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(1 to 10)</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at \&lt;console>:24</p> <p>（2）统计该RDD的条数</p> <p>scala> rdd.first</p> <p>res2: Int = 1</p> <h3 id=245-taken>2.4.5 take(n)案例</h3> <ol> <li> <p>作用：返回一个由RDD的前n个元素组成的数组</p> </li> <li> <p>需求：创建一个RDD，统计该RDD的条数</p> </li> </ol> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(Array(2,5,4,6,8,3))</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[2] at parallelize at \&lt;console>:24</p> <p>（2）统计该RDD的条数</p> <p>scala> rdd.take(3)</p> <p>res10: Array[Int] = Array(2, 5, 4)</p> <h3 id=246-takeorderedn>2.4.6 takeOrdered(n)案例</h3> <ol> <li> <p>作用：返回该RDD排序后的前n个元素组成的数组</p> </li> <li> <p>需求：创建一个RDD，统计该RDD的条数</p> </li> </ol> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(Array(2,5,4,6,8,3))</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[2] at parallelize at \&lt;console>:24</p> <p>（2）统计该RDD的条数</p> <p>scala> rdd.takeOrdered(3)</p> <p>res18: Array[Int] = Array(2, 3, 4)</p> <h3 id=247-aggregate>2.4.7 aggregate案例</h3> <ol> <li>参数：(zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U)</li> </ol> <p>2. 作用：aggregate函数将每个分区里面的元素通过seqOp和初始值进行聚合，然后用combine函数将每个分区的结果和初始值(zeroValue)进行combine操作。这个函数最终返回的类型不需要和RDD中元素类型一致。</p> <ol> <li>需求：创建一个RDD，将所有元素相加得到结果</li> </ol> <p>（1）创建一个RDD</p> <p>scala> var rdd1 = sc.makeRDD(1 to 10,2)</p> <p>rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[88] at makeRDD at \&lt;console>:24</p> <p>（2）将该RDD所有元素相加得到结果</p> <p>scala> rdd.aggregate(0)(<em>+</em>,<em>+</em>)</p> <p>res22: Int = 55</p> <h3 id=248-foldnumfunc>2.4.8 fold(num)(func)案例</h3> <ol> <li> <p>作用：折叠操作，aggregate的简化操作，seqop和combop一样。</p> </li> <li> <p>需求：创建一个RDD，将所有元素相加得到结果</p> </li> </ol> <p>（1）创建一个RDD</p> <p>scala> var rdd1 = sc.makeRDD(1 to 10,2)</p> <p>rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[88] at makeRDD at \&lt;console>:24</p> <p>（2）将该RDD所有元素相加得到结果</p> <p>scala> rdd.fold(0)(<em>+</em>)</p> <p>res24: Int = 55</p> <h3 id=249-saveastextfilepath>2.4.9 saveAsTextFile(path)</h3> <p>作用：将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本</p> <h3 id=2410-saveassequencefilepath>2.4.10 saveAsSequenceFile(path)</h3> <p>作用：将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下，可以使HDFS或者其他Hadoop支持的文件系统。</p> <h3 id=2411-saveasobjectfilepath>2.4.11 saveAsObjectFile(path)</h3> <p>作用：用于将RDD中的元素序列化成对象，存储到文件中。</p> <h3 id=2412-countbykey>2.4.12 countByKey()案例</h3> <ol> <li> <p>作用：针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数。</p> </li> <li> <p>需求：创建一个PairRDD，统计每种key的个数</p> </li> </ol> <p>（1）创建一个PairRDD</p> <p>scala> val rdd = sc.parallelize(List((1,3),(1,2),(1,4),(2,3),(3,6),(3,8)),3)</p> <p>rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[95] at parallelize at \&lt;console>:24</p> <p>（2）统计每种key的个数</p> <p>scala> rdd.countByKey</p> <p>res63: scala.collection.Map[Int,Long] = Map(3 -> 2, 1 -> 3, 2 -> 1)</p> <h3 id=2413-foreachfunc>2.4.13 foreach(func)案例</h3> <ol> <li> <p>作用：在数据集的每一个元素上，运行函数func进行更新。</p> </li> <li> <p>需求：创建一个RDD，对每个元素进行打印</p> </li> </ol> <p>（1）创建一个RDD</p> <p>scala> var rdd = sc.makeRDD(1 to 5,2)</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[107] at makeRDD at \&lt;console>:24</p> <p>（2）对该RDD每个元素进行打印</p> <p>scala> rdd.foreach(println(_))</p> <p>3</p> <p>4</p> <p>5</p> <p>1</p> <p>2</p> <h2 id=25-rdd>2.5 RDD中的函数传递</h2> <p>在实际开发中我们往往需要自己定义一些对于RDD的操作，那么此时需要主要的是，初始化工作是在Driver端进行的，而实际运行程序是在Executor端进行的，这就涉及到了跨进程通信，是需要序列化的。下面我们看几个例子：</p> <h3 id=251>2.5.1 传递一个方法</h3> <p>1．创建一个类</p> <p>class Search(s:String){</p> <p>//过滤出包含字符串的数据</p> <p>def isMatch(s: String): Boolean = {</p> <p>s.contains(query)</p> <p>}</p> <p>//过滤出包含字符串的RDD</p> <p>def getMatch1 (rdd: RDD[String]): RDD[String] = {</p> <p>rdd.filter(isMatch)</p> <p>}</p> <p>//过滤出包含字符串的RDD</p> <p>def getMatche2(rdd: RDD[String]): RDD[String] = {</p> <p>rdd.filter(x => x.contains(query))</p> <p>}</p> <p>}</p> <p>2．创建Spark主程序</p> <p>object SeriTest {</p> <p>def main(args: Array[String]): Unit = {</p> <p>//1.初始化配置信息及SparkContext</p> <p>val sparkConf: SparkConf = new SparkConf().setAppName(&ldquo;WordCount&rdquo;).setMaster(&ldquo;local[*]&rdquo;)</p> <p>val sc = new SparkContext(sparkConf)</p> <p>//2.创建一个RDD</p> <p>val rdd: RDD[String] = sc.parallelize(Array(&ldquo;hadoop&rdquo;, &ldquo;spark&rdquo;, &ldquo;hive&rdquo;, &ldquo;atguigu&rdquo;))</p> <p>//3.创建一个Search对象</p> <p>val search = new Search()</p> <p>//4.运用第一个过滤函数并打印结果</p> <p>val match1: RDD[String] = search.getMatche1(rdd)</p> <p>match1.collect().foreach(println)</p> <p>}</p> <p>}</p> <p>3．运行程序</p> <p>Exception in thread &ldquo;main&rdquo; org.apache.spark.SparkException: Task not serializable</p> <p>at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:298)</p> <p>at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:288)</p> <p>at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:108)</p> <p>at org.apache.spark.SparkContext.clean(SparkContext.scala:2101)</p> <p>at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:387)</p> <p>at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:386)</p> <p>at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)</p> <p>at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)</p> <p>at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)</p> <p>at org.apache.spark.rdd.RDD.filter(RDD.scala:386)</p> <p>at com.atguigu.Search.getMatche1(SeriTest.scala:39)</p> <p>at com.atguigu.SeriTest$.main(SeriTest.scala:18)</p> <p>at com.atguigu.SeriTest.main(SeriTest.scala)</p> <p>Caused by: java.io.NotSerializableException: com.atguigu.Search</p> <p>4．问题说明</p> <p>//过滤出包含字符串的RDD</p> <p>def getMatch1 (rdd: RDD[String]): RDD[String] = {</p> <p>rdd.filter(isMatch)</p> <p>}</p> <p>在这个方法中所调用的方法isMatch()是定义在Search这个类中的，实际上调用的是this. isMatch()，this表示Search这个类的对象，程序在运行过程中需要将Search对象序列化以后传递到Executor端。</p> <p>5．解决方案</p> <p>使类继承scala.Serializable即可。</p> <p>class Search() extends Serializable{&hellip;}</p> <h3 id=252>2.5.2 传递一个属性</h3> <p>1．创建Spark主程序</p> <p>object TransmitTest {</p> <p>def main(args: Array[String]): Unit = {</p> <p>//1.初始化配置信息及SparkContext</p> <p>val sparkConf: SparkConf = new SparkConf().setAppName(&ldquo;WordCount&rdquo;).setMaster(&ldquo;local[*]&rdquo;)</p> <p>val sc = new SparkContext(sparkConf)</p> <p>//2.创建一个RDD</p> <p>val rdd: RDD[String] = sc.parallelize(Array(&ldquo;hadoop&rdquo;, &ldquo;spark&rdquo;, &ldquo;hive&rdquo;, &ldquo;atguigu&rdquo;))</p> <p>//3.创建一个Search对象</p> <p>val search = new Search()</p> <p>//4.运用第一个过滤函数并打印结果</p> <p>val match1: RDD[String] = search.getMatche2(rdd)</p> <p>match1.collect().foreach(println)</p> <p>}</p> <p>}</p> <p>2．运行程序</p> <p>Exception in thread &ldquo;main&rdquo; org.apache.spark.SparkException: Task not serializable</p> <p>at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:298)</p> <p>at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:288)</p> <p>at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:108)</p> <p>at org.apache.spark.SparkContext.clean(SparkContext.scala:2101)</p> <p>at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:387)</p> <p>at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:386)</p> <p>at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)</p> <p>at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)</p> <p>at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)</p> <p>at org.apache.spark.rdd.RDD.filter(RDD.scala:386)</p> <p>at com.atguigu.Search.getMatche1(SeriTest.scala:39)</p> <p>at com.atguigu.SeriTest$.main(SeriTest.scala:18)</p> <p>at com.atguigu.SeriTest.main(SeriTest.scala)</p> <p>Caused by: java.io.NotSerializableException: com.atguigu.Search</p> <p>3．问题说明</p> <p>//过滤出包含字符串的RDD</p> <p>def getMatche2(rdd: RDD[String]): RDD[String] = {</p> <p>rdd.filter(x => x.contains(query))</p> <p>}</p> <p>在这个方法中所调用的方法query是定义在Search这个类中的字段，实际上调用的是this. query，this表示Search这个类的对象，程序在运行过程中需要将Search对象序列化以后传递到Executor端。</p> <p>4．解决方案</p> <p>1）使类继承scala.Serializable即可。</p> <p>class Search() extends Serializable{&hellip;}</p> <p>2）将类变量query赋值给局部变量</p> <p>修改getMatche2为</p> <p>//过滤出包含字符串的RDD</p> <p>def getMatche2(rdd: RDD[String]): RDD[String] = {</p> <p>val query_ : String = this.query//将类变量赋值给局部变量</p> <p>rdd.filter(x => x.contains(query_))</p> <p>}</p> <h2 id=26-rdd>2.6 RDD依赖关系</h2> <h3 id=261-lineage>2.6.1 Lineage</h3> <p>RDD只支持粗粒度转换，即在大量记录上执行的单个操作。将创建RDD的一系列Lineage（血统）记录下来，以便恢复丢失的分区。RDD的Lineage会记录RDD的元数据信息和转换行为，当该RDD的部分分区数据丢失时，它可以根据这些信息来重新运算和恢复丢失的数据分区。</p> <p><img alt src=d8e6c7a0ceb74ae137a8843d2bea28d3.png></p> <p>（1）读取一个HDFS文件并将其中内容映射成一个个元组</p> <p>scala> val wordAndOne = sc.textFile(&ldquo;/fruit.tsv&rdquo;).flatMap(<em>.split(&ldquo;\t&rdquo;)).map((</em>,1))</p> <p>wordAndOne: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[22] at map at \&lt;console>:24</p> <p>（2）统计每一种key对应的个数</p> <p>scala> val wordAndCount = wordAndOne.reduceByKey(<em>+</em>)</p> <p>wordAndCount: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[23] at reduceByKey at \&lt;console>:26</p> <p>（3）查看“wordAndOne”的Lineage</p> <p>scala> wordAndOne.toDebugString</p> <p>res5: String =</p> <p>(2) MapPartitionsRDD[22] at map at \&lt;console>:24 []</p> <p>| MapPartitionsRDD[21] at flatMap at \&lt;console>:24 []</p> <p>| /fruit.tsv MapPartitionsRDD[20] at textFile at \&lt;console>:24 []</p> <p>| /fruit.tsv HadoopRDD[19] at textFile at \&lt;console>:24 []</p> <p>（4）查看“wordAndCount”的Lineage</p> <p>scala> wordAndCount.toDebugString</p> <p>res6: String =</p> <p>(2) ShuffledRDD[23] at reduceByKey at \&lt;console>:26 []</p> <p>+-(2) MapPartitionsRDD[22] at map at \&lt;console>:24 []</p> <p>| MapPartitionsRDD[21] at flatMap at \&lt;console>:24 []</p> <p>| /fruit.tsv MapPartitionsRDD[20] at textFile at \&lt;console>:24 []</p> <p>| /fruit.tsv HadoopRDD[19] at textFile at \&lt;console>:24 []</p> <p>（5）查看“wordAndOne”的依赖类型</p> <p>scala> wordAndOne.dependencies</p> <p>res7: Seq[org.apache.spark.Dependency[_]] = List(org.apache.spark.OneToOneDependency\@5d5db92b)</p> <p>（6）查看“wordAndCount”的依赖类型</p> <p>scala> wordAndCount.dependencies</p> <p>res8: Seq[org.apache.spark.Dependency[_]] = List(org.apache.spark.ShuffleDependency\@63f3e6a8)</p> <p>注意：RDD和它依赖的父RDD（s）的关系有两种不同的类型，即窄依赖（narrow dependency）和宽依赖（wide dependency）。</p> <h3 id=262>2.6.2 窄依赖</h3> <p>窄依赖指的是每一个父RDD的Partition最多被子RDD的一个Partition使用,窄依赖我们形象的比喻为独生子女</p> <p><img alt=C:\Users\WHAlex\AppData\Local\Temp\1538964655(1).png src=47d11da031e6f2939dc75ceb451b0d15.png></p> <h3 id=263>2.6.3 宽依赖</h3> <p>宽依赖指的是多个子RDD的Partition会依赖同一个父RDD的Partition，会引起shuffle,总结：宽依赖我们形象的比喻为超生</p> <p><img alt=C:\Users\WHAlex\AppData\Local\Temp\1538964695(1).png src=53eab5e215e113315ae398cd249fd7be.png></p> <h3 id=264-dag>2.6.4 DAG</h3> <p>DAG(Directed Acyclic Graph)叫做有向无环图，原始的RDD通过一系列的转换就就形成了DAG，根据RDD之间的依赖关系的不同将DAG划分成不同的Stage，对于窄依赖，partition的转换处理在Stage中完成计算。对于宽依赖，由于有Shuffle的存在，只能在parent RDD处理完成后，才能开始接下来的计算，因此**宽依赖是划分Stage的依据**。</p> <p><img alt=图片1 src=735496c42a9bcb4dd2fa5544f6dabf81.png></p> <h3 id=265>2.6.5 任务划分（面试重点）</h3> <p>RDD任务切分中间分为：Application、Job、Stage和Task</p> <p>1）Application：初始化一个SparkContext即生成一个Application</p> <p>2）Job：一个Action算子就会生成一个Job</p> <p>3）Stage：根据RDD之间的依赖关系的不同将Job划分成不同的Stage，遇到一个宽依赖则划分一个Stage。</p> <p>4）Task：Stage是一个TaskSet，将Stage划分的结果发送到不同的Executor执行即为一个Task。</p> <p>注意：Application->Job->Stage-> Task每一层都是1对n的关系。</p> <h2 id=27-rdd>2.7 RDD缓存</h2> <p>RDD通过persist方法或cache方法可以将前面的计算结果缓存，默认情况下 persist() 会把数据以序列化的形式缓存在 JVM 的堆空间中。</p> <p>但是并不是这两个方法被调用时立即缓存，而是触发后面的action时，该RDD将会被缓存在计算节点的内存中，并供后面重用。</p> <p><img alt=QQ截图20151210132556 src=3d63cd8ac863c4eeee6824a841edd6ba.png></p> <p>通过查看源码发现cache最终也是调用了persist方法，默认的存储级别都是仅在内存存储一份，Spark的存储级别还有好多种，存储级别在object StorageLevel中定义的。</p> <p><img alt=QQ截图20151210140606 src=4fb41b571a83ff52dbd7bdf4eba9a4ed.png></p> <p>在存储级别的末尾加上“_2”来把持久化数据存为两份</p> <p><img alt src=146a9fe38b7b8af5ac5273b03e0961c8.png></p> <p>缓存有可能丢失，或者存储存储于内存的数据由于内存不足而被删除，RDD的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。通过基于RDD的一系列转换，丢失的数据会被重算，由于RDD的各个Partition是相对独立的，因此只需要计算丢失的部分即可，并不需要重算全部Partition。</p> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.makeRDD(Array(&ldquo;atguigu&rdquo;))</p> <p>rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[19] at makeRDD at \&lt;console>:25</p> <p>（2）将RDD转换为携带当前时间戳不做缓存</p> <p>scala> val nocache = rdd.map(_.toString+System.currentTimeMillis)</p> <p>nocache: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[20] at map at \&lt;console>:27</p> <p>（3）多次打印结果</p> <p>scala> nocache.collect</p> <p>res0: Array[String] = Array(atguigu1538978275359)</p> <p>scala> nocache.collect</p> <p>res1: Array[String] = Array(atguigu1538978282416)</p> <p>scala> nocache.collect</p> <p>res2: Array[String] = Array(atguigu1538978283199)</p> <p>（4）将RDD转换为携带当前时间戳并做缓存</p> <p>scala> val cache = rdd.map(_.toString+System.currentTimeMillis).cache</p> <p>cache: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[21] at map at \&lt;console>:27</p> <p>（5）多次打印做了缓存的结果</p> <p>scala> cache.collect</p> <p>res3: Array[String] = Array(atguigu1538978435705)</p> <p>scala> cache.collect</p> <p>res4: Array[String] = Array(atguigu1538978435705)</p> <p>scala> cache.collect</p> <p>res5: Array[String] = Array(atguigu1538978435705)</p> <h2 id=28-rdd-checkpoint>2.8 RDD CheckPoint</h2> <p>Spark中对于数据的保存除了持久化操作之外，还提供了一种检查点的机制，检查点（本质是通过将RDD写入Disk做检查点）是为了通过lineage做容错的辅助，lineage过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果之后有节点出现问题而丢失分区，从做检查点的RDD开始重做Lineage，就会减少开销。检查点通过将数据写入到HDFS文件系统实现了RDD的检查点功能。</p> <p>为当前RDD设置检查点。该函数将会创建一个二进制的文件，并存储到checkpoint目录中，该目录是用<a href=https://www.iteblog.com/archives/tag/spark/ >Spark</a>Context.setCheckpointDir()设置的。在checkpoint的过程中，该RDD的所有依赖于父RDD中的信息将全部被移除。对RDD进行checkpoint操作并不会马上被执行，必须执行Action操作才能触发。</p> <p>案例实操：</p> <p>（1）设置检查点</p> <p>scala> sc.setCheckpointDir(&ldquo;hdfs://hadoop102:9000/checkpoint&rdquo;)</p> <p>（2）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(Array(&ldquo;atguigu&rdquo;))</p> <p>rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[14] at parallelize at \&lt;console>:24</p> <p>（3）将RDD转换为携带当前时间戳并做checkpoint</p> <p>scala> val ch = rdd.map(_+System.currentTimeMillis)</p> <p>ch: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[16] at map at \&lt;console>:26</p> <p>scala> ch.checkpoint</p> <p>（4）多次打印结果</p> <p>scala> ch.collect</p> <p>res55: Array[String] = Array(atguigu1538981860336)</p> <p>scala> ch.collect</p> <p>res56: Array[String] = Array(atguigu1538981860504)</p> <p>scala> ch.collect</p> <p>res57: Array[String] = Array(atguigu1538981860504)</p> <p>scala> ch.collect</p> <p>res58: Array[String] = Array(atguigu1538981860504)</p> <h2 id=3-rdd>第3章 键值对RDD数据分区器</h2> <p>Spark目前支持Hash分区和Range分区，用户也可以自定义分区，Hash分区为当前的默认分区，Spark中分区器直接决定了RDD中分区的个数、RDD中每条数据经过Shuffle过程属于哪个分区和Reduce的个数</p> <p>注意：</p> <p>(1)只有Key-Value类型的RDD才有分区器的，非Key-Value类型的RDD分区器的值是None<br> (2)每个RDD的分区ID范围：0~numPartitions-1，决定这个值是属于那个分区的。</p> <h2 id=31-rdd>3.1 获取RDD分区</h2> <p>可以通过使用RDD的partitioner 属性来获取 RDD 的分区方式。它会返回一个 scala.Option 对象， 通过get方法获取其中的值。相关源码如下：</p> <p>def getPartition(key: Any): Int = key match {<br> case null => 0<br> case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)<br> }</p> <p>def nonNegativeMod(x: Int, mod: Int): Int = {<br> val rawMod = x % mod<br> rawMod + (if (rawMod \&lt; 0) mod else 0)<br> }</p> <p>（1）创建一个pairRDD</p> <p>scala> val pairs = sc.parallelize(List((1,1),(2,2),(3,3)))</p> <p>pairs: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[3] at parallelize at \&lt;console>:24</p> <p>（2）查看RDD的分区器</p> <p>scala> pairs.partitioner</p> <p>res1: Option[org.apache.spark.Partitioner] = None</p> <p>（3）导入HashPartitioner类</p> <p>scala> import org.apache.spark.HashPartitioner</p> <p>import org.apache.spark.HashPartitioner</p> <p>（4）使用HashPartitioner对RDD进行重新分区</p> <p>scala> val partitioned = pairs.partitionBy(new HashPartitioner(2))</p> <p>partitioned: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[4] at partitionBy at \&lt;console>:27</p> <p>（5）查看重新分区后RDD的分区器</p> <p>scala> partitioned.partitioner</p> <p>res2: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner\@2)</p> <h2 id=32-hash>3.2 Hash分区</h2> <p>HashPartitioner分区的原理：对于给定的key，计算其hashCode，并除以分区的个数取余，如果余数小于0，则用余数+分区的个数（否则加0），最后返回的值就是这个key所属的分区ID。</p> <p>使用Hash分区的实操</p> <p>scala> nopar.partitioner</p> <p>res20: Option[org.apache.spark.Partitioner] = None</p> <p>scala> val nopar = sc.parallelize(List((1,3),(1,2),(2,4),(2,3),(3,6),(3,8)),8)</p> <p>nopar: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[10] at parallelize at \&lt;console>:24</p> <p>scala>nopar.mapPartitionsWithIndex((index,iter)=>{ Iterator(index.toString+&rdquo; : &ldquo;+iter.mkString(&ldquo;|&ldquo;)) }).collect</p> <p>res0: Array[String] = Array(&ldquo;0 : &ldquo;, 1 : (1,3), 2 : (1,2), 3 : (2,4), &ldquo;4 : &ldquo;, 5 : (2,3), 6 : (3,6), 7 : (3,8))</p> <p>scala> val hashpar = nopar.partitionBy(new org.apache.spark.HashPartitioner(7))</p> <p>hashpar: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[12] at partitionBy at \&lt;console>:26</p> <p>scala> hashpar.count</p> <p>res18: Long = 6</p> <p>scala> hashpar.partitioner</p> <p>res21: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner\@7)</p> <p>scala> hashpar.mapPartitions(iter => Iterator(iter.length)).collect()</p> <p>res19: Array[Int] = Array(0, 3, 1, 2, 0, 0, 0)</p> <h2 id=33-ranger>3.3 Ranger分区</h2> <p>HashPartitioner分区弊端：可能导致每个分区中数据量的不均匀，极端情况下会导致某些分区拥有RDD的全部数据。</p> <p>RangePartitioner作用：将一定范围内的数映射到某一个分区内，尽量保证每个分区中数据量的均匀，而且分区与分区之间是有序的，一个分区中的元素肯定都是比另一个分区内的元素小或者大，但是分区内的元素是不能保证顺序的。简单的说就是将一定范围内的数映射到某一个分区内。实现过程为：</p> <p>第一步：先重整个RDD中抽取出样本数据，将样本数据排序，计算出每个分区的最大key值，形成一个Array[KEY]类型的数组变量rangeBounds；</p> <p>第二步：判断key在rangeBounds中所处的范围，给出该key值在下一个RDD中的分区id下标；该分区器要求RDD中的KEY类型必须是可以排序的</p> <h2 id=34>3.4 自定义分区</h2> <p>要实现自定义的分区器，你需要继承 org.apache.spark.Partitioner 类并实现下面三个方法。</p> <p>（1）numPartitions: Int:返回创建出来的分区数。</p> <p>（2）getPartition(key: Any): Int:返回给定键的分区编号(0到numPartitions-1)。</p> <p>（3）equals():Java 判断相等性的标准方法。这个方法的实现非常重要，Spark 需要用这个方法来检查你的分区器对象是否和其他分区器实例相同，这样 Spark 才可以判断两个 RDD 的分区方式是否相同。</p> <p>需求：将相同后缀的数据写入相同的文件，通过将相同后缀的数据分区到相同的分区并保存输出来实现。</p> <p>（1）创建一个pairRDD</p> <p>scala> val data = sc.parallelize(Array((1,1),(2,2),(3,3),(4,4),(5,5),(6,6)))</p> <p>data: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[3] at parallelize at \&lt;console>:24</p> <p>（2）定义一个自定义分区类</p> <p>scala> :paste</p> <p>// Entering paste mode (ctrl-D to finish)</p> <p>class CustomerPartitioner(numParts:Int) extends org.apache.spark.Partitioner{</p> <p>//覆盖分区数</p> <p>override def numPartitions: Int = numParts</p> <p>//覆盖分区号获取函数</p> <p>override def getPartition(key: Any): Int = {</p> <p>val ckey: String = key.toString</p> <p>ckey.substring(ckey.length-1).toInt%numParts</p> <p>}</p> <p>}</p> <p>// Exiting paste mode, now interpreting.</p> <p>defined class CustomerPartitioner</p> <p>（3）将RDD使用自定义的分区类进行重新分区</p> <p>scala> val par = data.partitionBy(new CustomerPartitioner(2))</p> <p>par: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[2] at partitionBy at \&lt;console>:27</p> <p>（4）查看重新分区后的数据分布</p> <p>scala> par.mapPartitionsWithIndex((index,items)=>items.map((index,_))).collect</p> <p>res3: Array[(Int, (Int, Int))] = Array((0,(2,2)), (0,(4,4)), (0,(6,6)), (1,(1,1)), (1,(3,3)), (1,(5,5)))</p> <p>使用自定义的 Partitioner 是很容易的:只要把它传给 partitionBy() 方法即可。Spark 中有许多依赖于数据混洗的方法，比如 join() 和 groupByKey()，它们也可以接收一个可选的 Partitioner 对象来控制输出数据的分区方式。</p> <h2 id=4>第4章 数据读取与保存</h2> <p>Spark的数据读取及数据保存可以从两个维度来作区分：文件格式以及文件系统。</p> <p>文件格式分为：Text文件、Json文件、Csv文件、Sequence文件以及Object文件；</p> <p>文件系统分为：本地文件系统、HDFS、HBASE以及数据库。</p> <h2 id=_1>文件类数据读取与保存</h2> <h3 id=411-text>4.1.1 Text文件</h3> <p>1）数据读取:textFile(String)</p> <p>scala> val hdfsFile = sc.textFile(&ldquo;hdfs://hadoop102:9000/fruit.txt&rdquo;)</p> <p>hdfsFile: org.apache.spark.rdd.RDD[String] = hdfs://hadoop102:9000/fruit.txt MapPartitionsRDD[21] at textFile at \&lt;console>:24</p> <p>2）数据保存: saveAsTextFile(String)</p> <p>scala> hdfsFile.saveAsTextFile(&ldquo;/fruitOut&rdquo;)</p> <p><img alt=C:\Users\WHAlex\AppData\Local\Temp\1538984947(1).png src=3e48b92a52107da0de7bcdab2f03ce0c.png></p> <h3 id=412-json>4.1.2 Json文件</h3> <p>如果JSON文件中每一行就是一个JSON记录，那么可以通过将JSON文件当做文本文件来读取，然后利用相关的JSON库对每一条数据进行JSON解析。</p> <p>注意：使用RDD读取JSON文件处理很复杂，同时SparkSQL集成了很好的处理JSON文件的方式，所以应用中多是采用SparkSQL处理JSON文件。</p> <p>（1）导入解析json所需的包</p> <p>scala> import scala.util.parsing.json.JSON</p> <p>（2）上传json文件到HDFS</p> <p>[atguigu\@hadoop102 spark]$ hadoop fs -put ./examples/src/main/resources/people.json /</p> <p>（3）读取文件</p> <p>scala> val json = sc.textFile(&ldquo;/people.json&rdquo;)</p> <p>json: org.apache.spark.rdd.RDD[String] = /people.json MapPartitionsRDD[8] at textFile at \&lt;console>:24</p> <p>（4）解析json数据</p> <p>scala> val result = json.map(JSON.parseFull)</p> <p>result: org.apache.spark.rdd.RDD[Option[Any]] = MapPartitionsRDD[10] at map at \&lt;console>:27</p> <p>（5）打印</p> <p>scala> result.collect</p> <p>res11: Array[Option[Any]] = Array(Some(Map(name -> Michael)), Some(Map(name -> Andy, age -> 30.0)), Some(Map(name -> Justin, age -> 19.0)))</p> <h3 id=413-sequence>4.1.3 Sequence文件</h3> <p>SequenceFile文件是<a href=http://lib.csdn.net/base/hadoop>Hadoop</a>用来存储二进制形式的key-value对而设计的一种平面文件(Flat File)。Spark 有专门用来读取 SequenceFile 的接口。在 SparkContext 中，可以调用 sequenceFile<a href=path> keyClass, valueClass</a>。</p> <p>注意：SequenceFile文件只针对PairRDD</p> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(Array((1,2),(3,4),(5,6)))</p> <p>rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[13] at parallelize at \&lt;console>:24</p> <p>（2）将RDD保存为Sequence文件</p> <p>scala> rdd.saveAsSequenceFile(&ldquo;file:///opt/module/spark/seqFile&rdquo;)</p> <p>（3）查看该文件</p> <p>[atguigu\@hadoop102 seqFile]$ pwd</p> <p>/opt/module/spark/seqFile</p> <p>[atguigu\@hadoop102 seqFile]$ ll</p> <p>总用量 8</p> <p>-rw-r&ndash;r&ndash; 1 atguigu atguigu 108 10月 9 10:29 part-00000</p> <p>-rw-r&ndash;r&ndash; 1 atguigu atguigu 124 10月 9 10:29 part-00001</p> <p>-rw-r&ndash;r&ndash; 1 atguigu atguigu 0 10月 9 10:29 _SUCCESS</p> <p>[atguigu\@hadoop102 seqFile]$ cat part-00000</p> <p>SEQ org.apache.hadoop.io.IntWritable org.apache.hadoop.io.IntWritableط</p> <p>（4）读取Sequence文件</p> <p>scala> val seq = sc.sequenceFile<a href title=file:///opt/module/spark/seqFile>Int,Int</a></p> <p>seq: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[18] at sequenceFile at \&lt;console>:24</p> <p>（5）打印读取后的Sequence文件</p> <p>scala> seq.collect</p> <p>res14: Array[(Int, Int)] = Array((1,2), (3,4), (5,6))</p> <h3 id=414>4.1.4 对象文件</h3> <p>对象文件是将对象序列化后保存的文件，采用Java的序列化机制。可以通过objectFile<a href=path>k,v</a> 函数接收一个路径，读取对象文件，返回对应的 RDD，也可以通过调用saveAsObjectFile() 实现对对象文件的输出。因为是序列化所以要指定类型。</p> <p>（1）创建一个RDD</p> <p>scala> val rdd = sc.parallelize(Array(1,2,3,4))</p> <p>rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[19] at parallelize at \&lt;console>:24</p> <p>（2）将RDD保存为Object文件</p> <p>scala> rdd.saveAsObjectFile(&ldquo;file:///opt/module/spark/objectFile&rdquo;)</p> <p>（3）查看该文件</p> <p>[atguigu\@hadoop102 objectFile]$ pwd</p> <p>/opt/module/spark/objectFile</p> <p>[atguigu\@hadoop102 objectFile]$ ll</p> <p>总用量 8</p> <p>-rw-r&ndash;r&ndash; 1 atguigu atguigu 142 10月 9 10:37 part-00000</p> <p>-rw-r&ndash;r&ndash; 1 atguigu atguigu 142 10月 9 10:37 part-00001</p> <p>-rw-r&ndash;r&ndash; 1 atguigu atguigu 0 10月 9 10:37 _SUCCESS</p> <p>[atguigu\@hadoop102 objectFile]$ cat part-00000</p> <p>SEQ!org.apache.hadoop.io.NullWritable&rdquo;org.apache.hadoop.io.BytesWritableW\@`l</p> <p>（4）读取Object文件</p> <p>scala> val objFile = sc.objectFile<a href title=file:///opt/module/spark/objectFile>Int</a></p> <p>objFile: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[31] at objectFile at \&lt;console>:24</p> <p>（5）打印读取后的Sequence文件</p> <p>scala> objFile.collect</p> <p>res19: Array[Int] = Array(1, 2, 3, 4)</p> <h2 id=_2>文件系统类数据读取与保存</h2> <h3 id=421-hdfs>4.2.1 HDFS</h3> <p>Spark的整个生态系统与Hadoop是完全兼容的,所以对于Hadoop所支持的文件类型或者数据库类型,Spark也同样支持.另外,由于Hadoop的API有新旧两个版本,所以Spark为了能够兼容Hadoop所有的版本,也提供了两套创建操作接口.对于外部存储创建操作而言,hadoopRDD和newHadoopRDD是最为抽象的两个函数接口,主要包含以下四个参数.</p> <p>1）输入格式(InputFormat): 制定数据输入的类型,如TextInputFormat等,新旧两个版本所引用的版本分别是org.apache.hadoop.mapred.InputFormat和org.apache.hadoop.mapreduce.InputFormat(NewInputFormat)</p> <p>2）键类型: 指定[K,V]键值对中K的类型</p> <p>3）值类型: 指定[K,V]键值对中V的类型</p> <p>4）分区值: 指定由外部存储生成的RDD的partition数量的最小值,如果没有指定,系统会使用默认值defaultMinSplits</p> <p>**注意:**其他创建操作的API接口都是为了方便最终的Spark程序开发者而设置的,是这两个接口的高效实现版本.例如,对于textFile而言,只有path这个指定文件路径的参数,其他参数在系统内部指定了默认值。</p> <p>1.在Hadoop中以压缩形式存储的数据,不需要指定解压方式就能够进行读取,因为Hadoop本身有一个解压器会根据压缩文件的后缀推断解压算法进行解压.</p> <p>2.如果用Spark从Hadoop中读取某种类型的数据不知道怎么读取的时候,上网查找一个使用map-reduce的时候是怎么读取这种这种数据的,然后再将对应的读取方式改写成上面的hadoopRDD和newAPIHadoopRDD两个类就行了</p> <h3 id=422-mysql>4.2.2 MySQL数据库连接</h3> <p>支持通过Java JDBC访问关系型数据库。需要通过JdbcRDD进行，示例如下:</p> <p>（1）添加依赖</p> <p>\&lt;dependency><br> \&lt;groupId>mysql\&lt;/groupId><br> \&lt;artifactId>mysql-connector-java\&lt;/artifactId><br> \&lt;version>5.1.27\&lt;/version><br> \&lt;/dependency></p> <p>（2）Mysql读取：</p> <p>package com.atguigu</p> <p>import java.sql.DriverManager</p> <p>import org.apache.spark.rdd.JdbcRDD</p> <p>import org.apache.spark.{SparkConf, SparkContext}</p> <p>object MysqlRDD {</p> <p>def main(args: Array[String]): Unit = {</p> <p>//1.创建spark配置信息</p> <p>val sparkConf: SparkConf = new SparkConf().setMaster(&ldquo;local[*]&rdquo;).setAppName(&ldquo;JdbcRDD&rdquo;)</p> <p>//2.创建SparkContext</p> <p>val sc = new SparkContext(sparkConf)</p> <p>//3.定义连接mysql的参数</p> <p>val driver = &ldquo;com.mysql.jdbc.Driver&rdquo;</p> <p>val url = &ldquo;jdbc:mysql://hadoop102:3306/rdd&rdquo;</p> <p>val userName = &ldquo;root&rdquo;</p> <p>val passWd = &ldquo;000000&rdquo;</p> <p>//创建JdbcRDD</p> <p>val rdd = new JdbcRDD(sc, () => {</p> <p>Class.forName(driver)</p> <p>DriverManager.getConnection(url, userName, passWd)</p> <p>},</p> <p>&ldquo;select * from `rddtable` where `id`>=?;&rdquo;,</p> <p>1,</p> <p>10,</p> <p>1,</p> <p>r => (r.getInt(1), r.getString(2))</p> <p>)</p> <p>//打印最后结果</p> <p>println(rdd.count())</p> <p>rdd.foreach(println)</p> <p>sc.stop()</p> <p>}</p> <p>}</p> <p>Mysql写入：</p> <p>def main(args: Array[String]) {<br> val sparkConf = new SparkConf().setMaster(&ldquo;local[2]&rdquo;).setAppName(&ldquo;HBaseApp&rdquo;)<br> val sc = new SparkContext(sparkConf)<br> val data = sc.parallelize(List(&ldquo;Female&rdquo;, &ldquo;Male&rdquo;,&rdquo;Female&rdquo;)) </p> <p>data.foreachPartition(insertData)<br> } </p> <p>def insertData(iterator: Iterator[String]): Unit = {</p> <p>Class.forName (&ldquo;com.mysql.jdbc.Driver&rdquo;).newInstance()<br> val conn = java.sql.DriverManager.getConnection(&ldquo;jdbc:mysql://hadoop102:3306/rdd&rdquo;, &ldquo;root&rdquo;, &ldquo;000000&rdquo;)<br> iterator.foreach(data => {<br> val ps = conn.prepareStatement(&ldquo;insert into rddtable(name) values (?)&rdquo;)<br> ps.setString(1, data)<br> ps.executeUpdate()<br> })<br> }</p> <h3 id=423-hbase>4.2.3 HBase数据库</h3> <p>由于 org.apache.hadoop.hbase.mapreduce.TableInputFormat 类的实现，Spark 可以通过Hadoop输入格式访问HBase。这个输入格式会返回键值对数据，其中键的类型为org. apache.hadoop.hbase.io.ImmutableBytesWritable，而值的类型为org.apache.hadoop.hbase.client.</p> <p>Result。</p> <p>（1）添加依赖</p> <p>\&lt;dependency></p> <p>\&lt;groupId>org.apache.hbase\&lt;/groupId></p> <p>\&lt;artifactId>hbase-server\&lt;/artifactId></p> <p>\&lt;version>1.3.1\&lt;/version></p> <p>\&lt;/dependency></p> <p>\&lt;dependency></p> <p>\&lt;groupId>org.apache.hbase\&lt;/groupId></p> <p>\&lt;artifactId>hbase-client\&lt;/artifactId></p> <p>\&lt;version>1.3.1\&lt;/version></p> <p>\&lt;/dependency></p> <p>（2）从HBase读取数据</p> <p>package com.atguigu</p> <p>import org.apache.hadoop.conf.Configuration</p> <p>import org.apache.hadoop.hbase.HBaseConfiguration</p> <p>import org.apache.hadoop.hbase.client.Result</p> <p>import org.apache.hadoop.hbase.io.ImmutableBytesWritable</p> <p>import org.apache.hadoop.hbase.mapreduce.TableInputFormat</p> <p>import org.apache.spark.rdd.RDD</p> <p>import org.apache.spark.{SparkConf, SparkContext}</p> <p>import org.apache.hadoop.hbase.util.Bytes</p> <p>object HBaseSpark {</p> <p>def main(args: Array[String]): Unit = {</p> <p>//创建spark配置信息</p> <p>val sparkConf: SparkConf = new SparkConf().setMaster(&ldquo;local[*]&rdquo;).setAppName(&ldquo;JdbcRDD&rdquo;)</p> <p>//创建SparkContext</p> <p>val sc = new SparkContext(sparkConf)</p> <p>//构建HBase配置信息</p> <p>val conf: Configuration = HBaseConfiguration.create()</p> <p>conf.set(&ldquo;hbase.zookeeper.quorum&rdquo;, &ldquo;hadoop102,hadoop103,hadoop104&rdquo;)</p> <p>conf.set(TableInputFormat.INPUT_TABLE, &ldquo;rddtable&rdquo;)</p> <p>//从HBase读取数据形成RDD</p> <p>val hbaseRDD: RDD[(ImmutableBytesWritable, Result)] = sc.<strong>newAPIHadoopRDD</strong>(</p> <p>conf,</p> <p>classOf[TableInputFormat],</p> <p>classOf[ImmutableBytesWritable],</p> <p>classOf[Result])</p> <p>val count: Long = hbaseRDD.count()</p> <p>println(count)</p> <p>//对hbaseRDD进行处理</p> <p>hbaseRDD.foreach {</p> <p>case (_, result) =></p> <p>val key: String = Bytes.toString(result.getRow)</p> <p>val name: String = Bytes.toString(result.getValue(Bytes.toBytes(&ldquo;info&rdquo;), Bytes.toBytes(&ldquo;name&rdquo;)))</p> <p>val color: String = Bytes.toString(result.getValue(Bytes.toBytes(&ldquo;info&rdquo;), Bytes.toBytes(&ldquo;color&rdquo;)))</p> <p>println(&ldquo;RowKey:&rdquo; + key + &ldquo;,Name:&rdquo; + name + &ldquo;,Color:&rdquo; + color)</p> <p>}</p> <p>//关闭连接</p> <p>sc.stop()</p> <p>}</p> <p>}</p> <p>3）往HBase写入</p> <p>def main(args: Array[String]) {</p> <p>//获取Spark配置信息并创建与spark的连接<br> val sparkConf = new SparkConf().setMaster(&ldquo;local[*]&rdquo;).setAppName(&ldquo;HBaseApp&rdquo;)<br> val sc = new SparkContext(sparkConf)</p> <p>//创建HBaseConf<br> val conf = HBaseConfiguration.create()<br> val jobConf = new JobConf(conf)<br> jobConf.setOutputFormat(classOf[TableOutputFormat])<br> jobConf.set(TableOutputFormat.OUTPUT_TABLE, &ldquo;fruit_spark&rdquo;)</p> <p>//构建Hbase表描述器</p> <p>val fruitTable = TableName.valueOf(&ldquo;fruit_spark&rdquo;)<br> val tableDescr = new HTableDescriptor(fruitTable)<br> tableDescr.addFamily(new HColumnDescriptor(&ldquo;info&rdquo;.getBytes))</p> <p>//创建Hbase表<br> val admin = new HBaseAdmin(conf)<br> if (admin.tableExists(fruitTable)) {<br> admin.disableTable(fruitTable)<br> admin.deleteTable(fruitTable)<br> }<br> admin.createTable(tableDescr)</p> <p>//定义往Hbase插入数据的方法<br> def convert(triple: (Int, String, Int)) = {<br> val put = new Put(Bytes.toBytes(triple._1))<br> put.addImmutable(Bytes.toBytes(&ldquo;info&rdquo;), Bytes.toBytes(&ldquo;name&rdquo;), Bytes.toBytes(triple._2))<br> put.addImmutable(Bytes.toBytes(&ldquo;info&rdquo;), Bytes.toBytes(&ldquo;price&rdquo;), Bytes.toBytes(triple._3))<br> (new ImmutableBytesWritable, put)<br> }</p> <p>//创建一个RDD<br> val initialRDD = sc.parallelize(List((1,&rdquo;apple&rdquo;,11), (2,&rdquo;banana&rdquo;,12), (3,&rdquo;pear&rdquo;,13)))</p> <p>//将RDD内容写到HBase<br> val localData = initialRDD.map(convert) </p> <p>localData.saveAsHadoopDataset(jobConf)<br> }</p> <h2 id=5-rdd>第5章 RDD编程进阶</h2> <h2 id=51>5.1 累加器</h2> <p>累加器用来对信息进行聚合，通常在向 Spark传递函数时，比如使用 map() 函数或者用 filter() 传条件时，可以使用驱动器程序中定义的变量，但是集群中运行的每个任务都会得到这些变量的一份新的副本，更新这些副本的值也不会影响驱动器中的对应变量。如果我们想实现所有分片处理时更新共享变量的功能，那么累加器可以实现我们想要的效果。</p> <h3 id=511>5.1.1 系统累加器</h3> <p>针对一个输入的日志文件，如果我们想计算文件中所有空行的数量，我们可以编写以下程序：</p> <p>scala> val notice = sc.textFile(&ldquo;./NOTICE&rdquo;)</p> <p>notice: org.apache.spark.rdd.RDD[String] = ./NOTICE MapPartitionsRDD[40] at textFile at \&lt;console>:32</p> <p>scala> val blanklines = sc.accumulator(0)</p> <p>warning: there were two deprecation warnings; re-run with -deprecation for details</p> <p>blanklines: org.apache.spark.Accumulator[Int] = 0</p> <p>scala> val tmp = notice.flatMap(line => {</p> <p>| if (line == &ldquo;&rdquo;) {</p> <p>| blanklines += 1</p> <p>| }</p> <p>| line.split(&rdquo; &ldquo;)</p> <p>| })</p> <p>tmp: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[41] at flatMap at \&lt;console>:36</p> <p>scala> tmp.count()</p> <p>res31: Long = 3213</p> <p>scala> blanklines.value</p> <p>res32: Int = 171</p> <p>累加器的用法如下所示。</p> <p>通过在驱动器中调用SparkContext.accumulator(initialValue)方法，创建出存有初始值的累加器。返回值为 org.apache.spark.Accumulator[T] 对象，其中 T 是初始值 initialValue 的类型。Spark闭包里的执行器代码可以使用累加器的 += 方法(在Java中是 add)增加累加器的值。 驱动器程序可以调用累加器的value属性(在Java中使用value()或setValue())来访问累加器的值。</p> <p>注意：工作节点上的任务不能访问累加器的值。从这些任务的角度来看，累加器是一个只写变量。</p> <p>对于要在行动操作中使用的累加器，Spark只会把每个任务对各累加器的修改应用一次。因此，如果想要一个无论在失败还是重复计算时都绝对可靠的累加器，我们必须把它放在 foreach() 这样的行动操作中。转化操作中累加器可能会发生不止一次更新</p> <h3 id=512>5.1.2 自定义累加器</h3> <p>自定义累加器类型的功能在1.X版本中就已经提供了，但是使用起来比较麻烦，在2.0版本后，累加器的易用性有了较大的改进，而且官方还提供了一个新的抽象类：AccumulatorV2来提供更加友好的自定义类型累加器的实现方式。实现自定义类型累加器需要继承AccumulatorV2并至少覆写下例中出现的方法，下面这个累加器可以用于在程序运行过程中收集一些文本类信息，最终以Set[String]的形式返回。1</p> <p>package com.atguigu.spark </p> <p>import org.apache.spark.util.AccumulatorV2<br> import org.apache.spark.{SparkConf, SparkContext}<br> import scala.collection.JavaConversions._ </p> <p>class LogAccumulator extends org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]] {<br> private val _logArray: java.util.Set[String] = new java.util.HashSet<a href>String</a> </p> <p>override def isZero: Boolean = {<br> _logArray.isEmpty<br> } </p> <p>override def reset(): Unit = {<br> _logArray.clear()<br> } </p> <p>override def add(v: String): Unit = {<br> _logArray.add(v)<br> } </p> <p>override def merge(other: org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]]): Unit = {<br> other match {<br> case o: LogAccumulator => _logArray.addAll(o.value)<br> } </p> <p>} </p> <p>override def value: java.util.Set[String] = {<br> java.util.Collections.unmodifiableSet(_logArray)<br> } </p> <p>override def copy():org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]] = {<br> val newAcc = new LogAccumulator()<br> _logArray.synchronized{<br> newAcc._logArray.addAll(_logArray)<br> }<br> newAcc<br> }<br> } </p> <p>// 过滤掉带字母的<br> object LogAccumulator {<br> def main(args: Array[String]) {<br> val conf=new SparkConf().setAppName(&ldquo;LogAccumulator&rdquo;)<br> val sc=new SparkContext(conf) </p> <p>val accum = new LogAccumulator<br> sc.register(accum, &ldquo;logAccum&rdquo;)<br> val sum = sc.parallelize(Array(&ldquo;1&rdquo;, &ldquo;2a&rdquo;, &ldquo;3&rdquo;, &ldquo;4b&rdquo;, &ldquo;5&rdquo;, &ldquo;6&rdquo;, &ldquo;7cd&rdquo;, &ldquo;8&rdquo;, &ldquo;9&rdquo;), 2).filter(line => {<br> val pattern = &ldquo;&rdquo;&ldquo;^-?(\d+)&rdquo;&ldquo;&rdquo;<br> val flag = line.matches(pattern)<br> if (!flag) {<br> accum.add(line)<br> }<br> flag<br> }).map(_.toInt).reduce(_ + _) </p> <p>println(&ldquo;sum: &rdquo; + sum)<br> for (v \&lt;- accum.value) print(v + &ldquo;&rdquo;)<br> println()<br> sc.stop()<br> }<br> }</p> <h2 id=52>5.2 广播变量（调优策略）</h2> <p>广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值，以供一个或多个Spark操作使用。比如，如果你的应用需要向所有节点发送一个较大的只读查询表，甚至是机器学习算法中的一个很大的特征向量，广播变量用起来都很顺手。 在多个并行操作中使用同一个变量，但是 Spark会为每个任务分别发送。</p> <p>scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))</p> <p>broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(35)</p> <p>scala> broadcastVar.value</p> <p>res33: Array[Int] = Array(1, 2, 3)</p> <p>使用广播变量的过程如下：</p> <p>(1) 通过对一个类型 T 的对象调用 SparkContext.broadcast 创建出一个 Broadcast[T] 对象。 任何可序列化的类型都可以这么实现。</p> <p>(2) 通过 value 属性访问该对象的值(在 Java 中为 value() 方法)。</p> <p>(3) 变量只会被发到各个节点一次，应作为只读值处理(修改这个值不会影响到别的节点)。</p> <h2 id=6>第6章 扩展</h2> <h2 id=61-rdd>6.1 RDD相关概念关系</h2> <p><img alt src=1583852161%281%29.jpg></p> <p>输入可能以多个文件的形式存储在HDFS上，每个File都包含了很多块，称为Block。当Spark读取这些文件作为输入时，会根据具体数据格式对应的InputFormat进行解析，一般是将若干个Block合并成一个输入分片，称为InputSplit，注意InputSplit不能跨越文件。随后将为这些输入分片生成具体的Task。InputSplit与Task是一一对应的关系。随后这些具体的Task每个都会被分配到集群上的某个节点的某个Executor去执行。</p> <ol> <li> <p>每个节点可以起一个或多个Executor。</p> </li> <li> <p>每个Executor由若干core组成，每个Executor的每个core一次只能执行一个Task。</p> </li> <li> <p>每个Task执行的结果就是生成了目标RDD的一个partiton。</p> </li> </ol> <p>注意: 这里的core是虚拟的core而不是机器的物理CPU核，可以理解为就是Executor的一个工作线程。而 Task被执行的并发度 = Executor数目 * 每个Executor核数。至于partition的数目：</p> <ol> <li> <p>对于数据读入阶段，例如sc.textFile，输入文件被划分为多少InputSplit就会需要多少初始Task。</p> </li> <li> <p>在Map阶段partition数目保持不变。</p> </li> <li> <p>在Reduce阶段，RDD的聚合会触发shuffle操作，聚合后的RDD的partition数目跟具体操作有关，例如repartition操作会聚合成指定分区数，还有一些算子是可配置的。</p> </li> </ol> <p>RDD在计算的时候，每个分区都会起一个task，所以rdd的分区数目决定了总的的task数目。申请的计算节点（Executor）数目和每个计算节点核数，决定了你同一时刻可以并行执行的task。</p> <p>比如的RDD有100个分区，那么计算的时候就会生成100个task，你的资源配置为10个计算节点，每个两2个核，同一时刻可以并行的task数目为20，计算这个RDD就需要5个轮次。如果计算资源不变，你有101个task的话，就需要6个轮次，在最后一轮中，只有一个task在执行，其余核都在空转。如果资源不变，你的RDD只有2个分区，那么同一时刻只有2个task运行，其余18个核空转，造成资源浪费。这就是在spark调优中，增大RDD分区数目，增大任务并行度的做法。</p> </article> </div> </div> </main> <footer class=md-footer> <div class=md-footer-nav> <nav class="md-footer-nav__inner md-grid" aria-label=Footer> <a href=../Spark%E5%9F%BA%E7%A1%80/ title=Spark基础 class="md-footer-nav__link md-footer-nav__link--prev" rel=prev> <div class="md-footer-nav__button md-icon"> <svg xmlns=http://www.w3.org/2000/svg viewbox="0 0 24 24"><path d="M20 11v2H8l5.5 5.5-1.42 1.42L4.16 12l7.92-7.92L13.5 5.5 8 11h12z"/></svg> </div> <div class=md-footer-nav__title> <div class=md-ellipsis> <span class=md-footer-nav__direction> 上一页 </span> Spark基础 </div> </div> </a> <a href=../SparkSql/ title=SparkSql class="md-footer-nav__link md-footer-nav__link--next" rel=next> <div class=md-footer-nav__title> <div class=md-ellipsis> <span class=md-footer-nav__direction> 下一页 </span> SparkSql </div> </div> <div class="md-footer-nav__button md-icon"> <svg xmlns=http://www.w3.org/2000/svg viewbox="0 0 24 24"><path d="M4 11v2h12l-5.5 5.5 1.42 1.42L19.84 12l-7.92-7.92L10.5 5.5 16 11H4z"/></svg> </div> </a> </nav> </div> <div class="md-footer-meta md-typeset"> <div class="md-footer-meta__inner md-grid"> <div class=md-footer-copyright> <div class=md-footer-copyright__highlight> Copyright &copy; 2018 - 2029 Dayet 296577630@qq.com </div> Made with <a href=https://squidfunk.github.io/mkdocs-material/ target=_blank rel=noopener> Material for MkDocs </a> </div> </div> </div> </footer> </div> <script src=../../assets/javascripts/vendor.2d1db4bd.min.js></script> <script src=../../assets/javascripts/bundle.6627ddf3.min.js></script><script id=__lang type=application/json>{"clipboard.copy": "\u590d\u5236", "clipboard.copied": "\u5df2\u590d\u5236", "search.config.lang": "ja", "search.config.pipeline": "trimmer, stemmer", "search.config.separator": "[\\uff0c\\u3002]+", "search.result.placeholder": "\u952e\u5165\u4ee5\u5f00\u59cb\u641c\u7d22", "search.result.none": "\u6ca1\u6709\u627e\u5230\u7b26\u5408\u6761\u4ef6\u7684\u7ed3\u679c", "search.result.one": "\u627e\u5230 1 \u4e2a\u7b26\u5408\u6761\u4ef6\u7684\u7ed3\u679c", "search.result.other": "# \u4e2a\u7b26\u5408\u6761\u4ef6\u7684\u7ed3\u679c"}</script> <script>
        app = initialize({
          base: "../..",
          features: [],
          search: Object.assign({
            worker: "../../assets/javascripts/worker/search.5eca75d3.min.js"
          }, typeof search !== "undefined" && search)
        })
      </script> </body> </html>